langgraph_inmem_server_local_testing_with_sdk_client.py

1import asyncio
2from langgraph.builder import StateGraph, START
3from langgraph_sdk import get_client
4from langgraph_runtime_inmem import InMemServer
5
6# 1. Define a simple graph
7def my_node(state: dict):
8    return {"messages": state.get("messages", []) + ["Hello from the graph!"]}
9
10builder = StateGraph(dict)
11builder.add_node("my_node", my_node)
12builder.add_edge(START, "my_node")
13graph = builder.compile()
14
15async def main():
16    # 2. Start the in-memory server with your graph
17    # This simulates the LangGraph Cloud environment locally
18    async with InMemServer(graphs={"my_graph": graph}) as server:
19        # 3. Use the SDK client to interact with the local server
20        client = get_client(url=server.url)
21        
22        # 4. Create a thread and run the graph
23        thread = await client.threads.create()
24        input_data = {"messages": ["Hi there!"]}
25        
26        print("Running graph...")
27        async for event in client.runs.stream(
28            thread["thread_id"],
29            "my_graph",
30            input=input_data
31        ):
32            print(event)
33
34if __name__ == "__main__":
35    asyncio.run(main())