Implementation:Langchain ai Langgraph RemoteGraph Init
| Property | Value |
|---|---|
| API | `RemoteGraph.__init__` and `get_client` |
| Type | API Doc |
| Source | `libs/langgraph/langgraph/pregel/remote.py` and `libs/sdk-py/langgraph_sdk/client.py` |
| Library | langgraph (RemoteGraph), langgraph-sdk (get_client) |
| Related Workflow | CLI_Deployment |
Overview
The `RemoteGraph` class provides a graph-compatible client for interacting with deployed LangGraph API servers. Its `__init__` method configures the connection to a remote server, creating both async and sync HTTP clients. The `get_client` function from `langgraph_sdk` creates a standalone `LangGraphClient` for direct API access without the graph protocol abstraction. Together they provide the primary interfaces for connecting client applications to deployed LangGraph services.
Description
RemoteGraph.__init__
The `RemoteGraph` constructor (at `libs/langgraph/langgraph/pregel/remote.py:L122-169`) initializes a remote graph proxy:
- Sets the `assistant_id` (positional-only parameter) identifying which graph to interact with on the server.
- Sets the `name` attribute (defaults to `assistant_id` if not provided), used when adding `RemoteGraph` as a subgraph node.
- Stores the optional `config` for merging with per-call configurations.
- Stores the `distributed_tracing` flag for LangSmith header injection.
- Creates an async `LangGraphClient` via `get_client(url, api_key, headers)` if `url` is provided and no explicit `client` is given.
- Creates a sync `SyncLangGraphClient` via `get_sync_client(url, api_key, headers)` if `url` is provided and no explicit `sync_client` is given.
At least one of `url`, `client`, or `sync_client` must be provided. If only `client` is provided, synchronous methods will raise `ValueError`. If only `sync_client` is provided, asynchronous methods will raise `ValueError`.
get_client
The `get_client` function (at `libs/sdk-py/langgraph_sdk/client.py:L178`) creates and configures a `LangGraphClient`:
- URL handling: If `url` is `None`, attempts in-process ASGI transport connection to a co-located server. Falls back to deferred transport registration.
- Authentication: Resolves the API key from the explicit parameter or environment variables (`LANGGRAPH_API_KEY`, `LANGSMITH_API_KEY`, `LANGCHAIN_API_KEY`).
- HTTP client creation: Creates an `httpx.AsyncClient` with:
- Retry transport (5 retries) for remote connections.
- Configurable timeout (default: connect=5s, read=300s, write=300s, pool=5s).
- Authentication and custom headers.
- Client wrapping: Wraps the HTTP client in a `LangGraphClient` which exposes sub-clients for assistants, threads, runs, crons, and store.
LangGraphClient Structure
The `LangGraphClient` returned by `get_client` exposes:
| Sub-client | Purpose |
|---|---|
| `client.assistants` | Manage graph versions, schemas, and configurations |
| `client.threads` | Create, list, and manage conversation threads |
| `client.runs` | Execute graph invocations, stream results, manage run lifecycle |
| `client.crons` | Schedule recurring graph executions |
| `client.store` | Persistent key-value storage operations |
Code Reference
Source Location
| Item | File | Line |
|---|---|---|
| `RemoteGraph.__init__` | `libs/langgraph/langgraph/pregel/remote.py` | L122-169 |
| `RemoteGraph` class | `libs/langgraph/langgraph/pregel/remote.py` | L108 |
| `get_client` | `libs/sdk-py/langgraph_sdk/client.py` | L178 |
| `LangGraphClient` | `libs/sdk-py/langgraph_sdk/client.py` | L292 |
Signature
class RemoteGraph(PregelProtocol):
def __init__(
self,
assistant_id: str, # positional-only
/,
*,
url: str | None = None,
api_key: str | None = None,
headers: dict[str, str] | None = None,
client: LangGraphClient | None = None,
sync_client: SyncLangGraphClient | None = None,
config: RunnableConfig | None = None,
name: str | None = None,
distributed_tracing: bool = False,
):
"""Specify url, api_key, and/or headers to create default sync and async clients.
If client or sync_client are provided, they will be used instead of the default
clients. At least one of url, client, or sync_client must be provided.
"""
...
def get_client(
*,
url: str | None = None,
api_key: str | None = NOT_PROVIDED,
headers: Mapping[str, str] | None = None,
timeout: TimeoutTypes | None = None,
) -> LangGraphClient:
"""Create and configure a LangGraphClient."""
...
Import
from langgraph.pregel.remote import RemoteGraph
from langgraph_sdk import get_client
I/O Contract
RemoteGraph.__init__
| Direction | Name | Type | Description |
|---|---|---|---|
| Input | `assistant_id` | `str` | Graph ID or assistant name on the remote server (positional-only) |
| Input | `url` | None` | Base URL of the LangGraph API server |
| Input | `api_key` | None` | API key for authentication |
| Input | `headers` | None` | Additional HTTP headers |
| Input | `client` | None` | Pre-configured async client |
| Input | `sync_client` | None` | Pre-configured sync client |
| Input | `config` | None` | Default configuration merged with per-call configs |
| Input | `name` | None` | Human-readable name (defaults to `assistant_id`) |
| Input | `distributed_tracing` | `bool` | Enable LangSmith distributed tracing headers (default `False`) |
| Output | instance | `RemoteGraph` | Initialized remote graph proxy |
get_client
| Direction | Name | Type | Description |
|---|---|---|---|
| Input | `url` | None` | Base URL of the API; `None` for in-process connection |
| Input | `api_key` | None` | API key; auto-loaded from env if not provided |
| Input | `headers` | None` | Additional HTTP headers |
| Input | `timeout` | None` | HTTP timeout (default: connect=5s, read=300s, write=300s, pool=5s) |
| Output | return | `LangGraphClient` | Configured async client with sub-clients |
Usage Examples
Basic RemoteGraph Connection
from langgraph.pregel.remote import RemoteGraph
# Connect to a deployed LangGraph server
remote_graph = RemoteGraph(
"my_agent",
url="http://localhost:8123",
)
# Invoke like a local graph
result = remote_graph.invoke(
{"messages": [{"role": "user", "content": "Hello!"}]},
config={"configurable": {"thread_id": "thread-123"}},
)
RemoteGraph with Distributed Tracing
from langgraph.pregel.remote import RemoteGraph
remote = RemoteGraph(
"my_agent",
url="https://my-deployment.langchain.com",
api_key="lsv2_...",
distributed_tracing=True,
)
# Tracing headers are automatically included in requests
for chunk in remote.stream(
{"messages": [{"role": "user", "content": "Analyze this data"}]},
stream_mode="updates",
):
print(chunk)
RemoteGraph as a Subgraph Node
from langgraph.graph import StateGraph
from langgraph.pregel.remote import RemoteGraph
# Create a remote graph for a deployed agent
research_agent = RemoteGraph(
"research_agent",
url="http://research-service:8123",
name="researcher",
)
# Use it as a node in a local graph
builder = StateGraph(dict)
builder.add_node("researcher", research_agent)
Using get_client for Direct API Access
from langgraph_sdk import get_client
# Create an async client
client = get_client(url="http://localhost:8123")
# List available assistants
assistants = await client.assistants.search()
# Create a thread and run a graph
thread = await client.threads.create()
result = await client.runs.wait(
thread_id=thread["thread_id"],
assistant_id="my_agent",
input={"messages": [{"role": "user", "content": "Hello!"}]},
)
print(result)
In-Process Client (Server-Side)
from langgraph_sdk import get_client
# Connect in-process (when running inside the LangGraph server)
client = get_client()
# Call another graph within the same server
async def my_node(state):
result = await client.runs.wait(
thread_id=None,
assistant_id="helper_agent",
input={"query": state["query"]},
)
return {"answer": result}