Connecting to a Deployment

1Two connection methods: SDK vs RemoteGraph

After LangGraph Platform is deployed and running, you can connect from the client in two ways:

Method 1: LangGraph SDK (recommended)

from langgraph_sdk import get_client

# 连接到本地部署（CLI 方式）
url_for_cli_deployment = "http://localhost:8123"
client = get_client(url=url_for_cli_deployment)

# 或连接到 LangGraph Cloud 托管部署
url_for_cloud = "https://langchain-academy-xxx.langgraph.app"
client = get_client(url=url_for_cloud)

SDK provides completeRuns、Threads、StoreThe management interface is the standard way of interacting with the LangGraph Server API.

Method 2: RemoteGraph (suitable for internal calls of LangGraph)

from langgraph.pregel.remote import RemoteGraph

# RemoteGraph 可以像本地图一样调用，适合在另一个图中嵌套远程图
graph_name = "task_maistro"
remote_graph = RemoteGraph(graph_name, url="http://localhost:8123")

# 用法与本地图完全相同：invoke / stream / get_state 等
result = await remote_graph.ainvoke({"messages": [...]}, config={...})

2Three core concepts: Runs, Threads, Store

3Runs: three execution modes

Mode 1: Running in the background (Fire and Forget)

# 创建线程
thread = await client.threads.create()

# 启动后台运行（立即返回，不等待结果）
config = {"configurable": {"user_id": "Test"}}
run = await client.runs.create(
    thread["thread_id"],
    "task_maistro",
    input={"messages": [HumanMessage(content="Add a ToDo to book travel to HK")]},
    config=config
)
# run["status"] 为 "pending"，继续异步执行中

Mode 2: Blocking and waiting (Blocking)

# 等待 run 完成（阻塞当前协程）
await client.runs.join(thread["thread_id"], run["run_id"])

# 此时 status 变为 "success"
result = await client.runs.get(thread["thread_id"], run["run_id"])
print(result["status"])  # "success"

Mode 3: Streaming output (Streaming)

# stream_mode="messages-tuple" → 逐 token 流式返回
async for chunk in client.runs.stream(
    thread["thread_id"],
    "task_maistro",
    input={"messages": [HumanMessage(content="What ToDo should I focus on first?")]},
    config=config,
    stream_mode="messages-tuple"
):
    if chunk.event == "messages":
        # 实时打印 AI 生成的每个 token
        print("".join(
            data_item['content'] for data_item in chunk.data
            if 'content' in data_item
        ), end="", flush=True)

4Threads: multiple rounds of interaction and state management

Get thread status

# 获取线程的当前状态（最新 checkpoint）
thread_state = await client.threads.get_state(thread['thread_id'])

# 访问消息历史
from langchain_core.messages import convert_to_messages
for m in convert_to_messages(thread_state['values']['messages']):
    m.pretty_print()

# 获取完整历史（所有 checkpoints）
states = await client.threads.get_history(thread['thread_id'])

Copy thread (Fork)

# 复制线程：保留历史，但创建独立的新线程（不影响原线程）
copied_thread = await client.threads.copy(thread['thread_id'])

# 复制后的线程包含完整的历史消息
copied_state = await client.threads.get_state(copied_thread['thread_id'])

5Threads: Forking and Human-in-the-Loop

passget_history + update_state + checkpoint_id, which enables time travel and manual intervention in production environments.

# Step 1：获取线程历史，找到要分叉的 checkpoint
states = await client.threads.get_history(thread['thread_id'])
to_fork = states[-2]  # 选择倒数第二个 checkpoint

# 查看该 checkpoint 的关键信息
to_fork['checkpoint_id']   # '1efa2c00-6609-67ff-8000-491b1dcf8129'
to_fork['next']            # ['task_mAIstro'] — 下一步执行的节点

# Step 2：修改该 checkpoint 的状态（覆写同一消息 ID，而非追加）
forked_input = {
    "messages": HumanMessage(
        content="Give me a summary of all ToDos that need to be done in the next week.",
        id=to_fork['values']['messages'][0]['id']  # 复用原消息 ID → 覆写而非追加
    )
}
forked_config = await client.threads.update_state(
    thread["thread_id"],
    forked_input,
    checkpoint_id=to_fork['checkpoint_id']
)

# Step 3：从新 checkpoint 恢复执行（input=None 表示使用 checkpoint 中的状态）
async for chunk in client.runs.stream(
    thread["thread_id"],
    graph_name,
    input=None,
    config=config,
    checkpoint_id=forked_config['checkpoint_id'],  # 指定从哪个 checkpoint 开始
    stream_mode="messages-tuple"
):
    ...

6Store: CRUD operations across thread memory

SDK providesclient.storeInterface that allows direct manipulation of the long-term memory in the Store outside the graph, and inside the graphstore.put/search/getThe calling effect is the same.

search_items: Search by namespace

# task_maistro 将 ToDo 存储在 ("todo", "general", user_id) 命名空间
items = await client.store.search_items(
    ("todo", "general", "Test"),
    limit=5,
    offset=0
)

# 返回示例
# [{'value': {'task': 'Finish booking travel to Hong Kong',
#             'status': 'not started', 'deadline': '2024-11-22T23:59:59',
#             'solutions': [...], 'time_to_complete': 120},
#   'key': '18524803-c182-49de-9b10-08ccb0a06843',
#   'namespace': ['todo', 'general', 'Test'],
#   'created_at': '2024-11-14T19:37:41.664827+00:00', ...}]

put_item: write directly to memory

from uuid import uuid4

# 在图外直接写入 Store（不经过图逻辑）
await client.store.put_item(
    ("testing", "Test"),         # namespace
    key=str(uuid4()),               # 唯一 key
    value={"todo": "Test SDK put_item"}  # value 必须是字典
)

delete_item: delete memory

# 按 namespace + key 精确删除
await client.store.delete_item(
    ("testing", "Test"),
    key="3de441ba-8c79-4beb-8f52-00e4dcba68d4"
)

Namespace specification (convention of task_mAIstro)