Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Langchain ai Langgraph RemoteGraph Init

From Leeroopedia
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:

  1. Sets the `assistant_id` (positional-only parameter) identifying which graph to interact with on the server.
  2. Sets the `name` attribute (defaults to `assistant_id` if not provided), used when adding `RemoteGraph` as a subgraph node.
  3. Stores the optional `config` for merging with per-call configurations.
  4. Stores the `distributed_tracing` flag for LangSmith header injection.
  5. Creates an async `LangGraphClient` via `get_client(url, api_key, headers)` if `url` is provided and no explicit `client` is given.
  6. 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`:

  1. URL handling: If `url` is `None`, attempts in-process ASGI transport connection to a co-located server. Falls back to deferred transport registration.
  2. Authentication: Resolves the API key from the explicit parameter or environment variables (`LANGGRAPH_API_KEY`, `LANGSMITH_API_KEY`, `LANGCHAIN_API_KEY`).
  3. 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.
  4. 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}

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment