executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
sdk-clientsUse LangGraph Python SDK clients, API resource clients, auth headers, and thread-centric streaming v3; understand JS SDK status.
Independent GitHub snapshot · Automated classification · Original documentation governs
Use this sub-skill when an agent needs to connect Python code to a running LangGraph API server, manage API resources, or convert polling/run streaming code to thread-centric streaming v3.
get_client() or get_sync_client().client.threads.stream() with typed projections such as messages, tool_calls, values, output, subgraphs, and extensions.For graph authoring and invocation semantics, use graph-runtime. For prebuilt agent helpers that produce assistant graphs, use prebuilt-agents. For checkpoint and store persistence details, use persistence. For starting or deploying the API server, use cli-deployment when present.
Install the Python SDK in the target project environment:
python -m pip install langgraph-sdk
Check that the expected exports are present without contacting a server:
python skills/langgraph/sub-skills/sdk-clients/scripts/inspect_sdk_exports.py --json
Expected result: JSON with ok: true, callable get_client and get_sync_client, async/sync top-level client classes, resource client exports, and the auth environment fallback order.
The public factories are:
from langgraph_sdk import get_client, get_sync_client
async_client = get_client(
url="http://localhost:8123",
api_key=None,
headers={"X-Request-Source": "agent"},
timeout=30.0,
)
sync_client = get_sync_client(url="http://localhost:8123", api_key=None)
Parameter notes:
url is the LangGraph API base URL. get_sync_client() defaults to http://localhost:8123 when omitted.get_client(url=None) is async and first attempts an in-process server transport when used inside a LangGraph server; otherwise give an explicit URL for normal scripts.api_key accepts a string, None, or the default sentinel. A string becomes the bearer credential; None disables environment key lookup.api_key is omitted, keys auto-load in order: LANGGRAPH_API_KEY, then LANGSMITH_API_KEY, then LANGCHAIN_API_KEY.headers merges additional HTTP headers with auth headers. Do not hard-code secrets in source; prefer environment variables.timeout accepts an httpx.Timeout, a float total timeout, or a (connect, read, write, pool) tuple. Defaults are connect 5s, read 300s, write 300s, pool 5s.Use context managers in long-running apps so the underlying HTTP client closes cleanly:
async with get_client(url=server_url, api_key=token) as client:
assistants = await client.assistants.search()
with get_sync_client(url=server_url, api_key=token) as client:
assistants = client.assistants.search()
More client details: references/api-clients.md.
Top-level clients expose these resources:
client.assistants for graph assistant configuration and lookup.client.threads for conversational threads, state, history, and thread-centric streaming.client.runs for starting, waiting, joining, cancelling, and streaming runs.client.crons for scheduled runs.client.store for persistent document store operations.client.http for lower-level request helpers when a resource method is not enough.Keep async and sync styles separate. Await async resource calls from get_client(). Do not await sync calls from get_sync_client().
Prefer client.threads.stream() when a workflow needs one thread session with multiple typed views over the same run. The call returns a context manager. assistant_id is required. If thread_id is omitted, the SDK mints a UUIDv4 client-side and the server creates the thread lazily on first run.start.
import asyncio
from langgraph_sdk import get_client
client = get_client(url="http://localhost:8123", api_key=None)
async with client.threads.stream(assistant_id="agent") as thread:
await thread.run.start(input={"messages": [{"role": "user", "content": "hi"}]})
async def collect_messages():
return [stream async for stream in thread.messages]
async def collect_tools():
return [call async for call in thread.tool_calls]
messages, tool_calls = await asyncio.gather(collect_messages(), collect_tools())
final_values = await thread.output
Important streaming rules:
async with for async streams and with for sync streams. Calling run.start() or subscribing outside the context manager raises a runtime error.headers passed to threads.stream() are forwarded to command and event requests for that stream session.transport="sse" is the default. Async thread streaming can use transport="websocket" when the websockets optional dependency is installed and the server supports it.thread.extensions[name] access creates a subscription. Assign it once and reuse it inside the session to avoid duplicate extension subscriptions.More streaming patterns and conversion guidance: references/thread-streaming.md.
The JavaScript/TypeScript SDK content has moved out of this repository to the public LangGraph.js repository. For JS/TS clients, use LangGraph.js and the LangGraph.js docs. Do not infer current JS SDK APIs from this repository beyond that migration notice.
python -m pip show langgraph-sdk succeeds in the environment that will run the client code.python skills/langgraph/sub-skills/sdk-clients/scripts/inspect_sdk_exports.py reports OK: langgraph-sdk client exports available.api_key=None unless the local server actually requires auth.LANGGRAPH_API_KEY or a secret manager, not literals committed to source.client.assistants, client.threads, client.runs, client.crons, and client.store methods; sync code does not.run.start() and consumes required projections before exiting.