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 ServerRuntime

From Leeroopedia
Attribute Value
Source `libs/sdk-py/langgraph_sdk/runtime.py` (238 lines)
Domain SDK, Runtime
Principle Server_Runtime_Context
Library sdk-py
Status Beta
Import `from langgraph_sdk.runtime import ServerRuntime, AccessContext`

Overview

The `runtime.py` module in the `langgraph_sdk` package defines the server-side runtime context classes that are passed to graph builder factories within the LangGraph Agent Server. `ServerRuntime` is a type alias for the union of `_ExecutionRuntime` and `_ReadRuntime`, providing access to the authenticated user, store, access context, and (during execution) the graph run context.

Description

When the LangGraph Agent Server calls a graph factory, it passes a `ServerRuntime` instance that tells the factory why the graph is being constructed and provides runtime resources. The server accesses graphs in several contexts beyond just executing runs, and the factory must return a graph with the same topology in all cases.

AccessContext

`AccessContext` -- A `Literal` type alias defining the four contexts in which the server accesses graphs:

Value Category Description
`"threads.create_run"` Write Full graph execution via `graph.astream`. Context is available.
`"threads.update"` Write State update via `graph.aupdate_state`. Does not execute nodes or edges; only applies channel writers.
`"threads.read"` Read State inspection via `graph.aget_state` / `graph.aget_state_history`. Graph structure determines pending tasks.
`"assistants.read"` Introspection Schema/visualization via `graph.aget_graph`, `graph.aget_subgraphs`, `graph.aget_schemas`. Used by Studio UI and protocol integrations (MCP, A2A).

_ServerRuntimeBase

`_ServerRuntimeBase(Generic[ContextT])` -- Frozen, slot-based dataclass serving as the base for both runtime variants:

  • `access_context: AccessContext` -- Why the graph factory is being called.
  • `user: BaseUser | None` -- The authenticated user, or `None` if no custom auth is configured.
  • `store: BaseStore` -- Store for persistence and memory.
  • `execution_runtime` property -- Narrows to `_ExecutionRuntime` if the current instance is an execution runtime, otherwise returns `None`.
  • `ensure_user() -> BaseUser` -- Returns the authenticated user or raises `PermissionError` if not available.

_ExecutionRuntime

`_ExecutionRuntime(_ServerRuntimeBase[ContextT])` -- Runtime for `threads.create_run` context where the graph will be fully executed. Adds:

  • `context: ContextT` -- The graph run context, typed by the graph's `context_schema`. Only available during execution.

_ReadRuntime

`_ReadRuntime(_ServerRuntimeBase[ContextT])` -- Runtime for non-execution access contexts (introspection, state reads, state updates). Does not provide `context`.

ServerRuntime Type Alias

`ServerRuntime` -- A `TypeAliasType` defined as `_ExecutionRuntime[ContextT] | _ReadRuntime[ContextT]`. This is the type that graph factory functions should annotate their parameter with. Requires Agent Server version 0.7.30 or later.

Usage

import contextlib
from langgraph_sdk.runtime import ServerRuntime

@contextlib.asynccontextmanager
async def my_factory(runtime: ServerRuntime[MyCtx]):
    if ert := runtime.execution_runtime:
        # Only set up expensive resources during execution
        tools = await connect_tools(ert.context)
        yield create_graph(tools=tools)
        await disconnect_tools()
    else:
        yield create_graph(tools=[])

Code Reference

AccessContext

AccessContext = Literal[
    "threads.create_run",
    "threads.update",
    "threads.read",
    "assistants.read",
]

_ServerRuntimeBase Attributes

Attribute Type Description
`access_context` `AccessContext` Why the graph factory is being called.
`user` None` Authenticated user, or `None` without custom auth.
`store` `BaseStore` Store for persistence and memory.

_ServerRuntimeBase Methods

Method Signature Description
`execution_runtime` None` Narrow to execution runtime, or `None` if not in execution context.
`ensure_user` `() -> BaseUser` Return authenticated user or raise `PermissionError`.

_ExecutionRuntime Attributes

Attribute Type Default Description
`context` `ContextT` `None` Graph run context typed by `context_schema`.

ServerRuntime Type Alias

Name Definition Description
`ServerRuntime` _ReadRuntime[ContextT]` Union type passed to graph factory functions.

I/O Contract

Aspect Detail
Input Constructed by the LangGraph Agent Server and passed to graph factory functions.
Output Provides `access_context`, `user`, `store`, and (during execution) `context` to the factory.
Side Effects `ensure_user()` raises `PermissionError` if no custom auth is configured.
Thread Safety Instances are frozen dataclasses with slots, making them immutable and safe to share.
Server Version Requires Agent Server version 0.7.30 or later.

Usage Examples

Conditional Resource Setup

import contextlib
from dataclasses import dataclass
from langgraph_sdk.runtime import ServerRuntime

@dataclass
class MyCtx:
    mcp_endpoint: str

@contextlib.asynccontextmanager
async def my_factory(runtime: ServerRuntime[MyCtx]):
    if ert := runtime.execution_runtime:
        user_id = runtime.ensure_user().identity
        mcp_tools = await connect_mcp(ert.context.mcp_endpoint, user_id)
        yield create_agent("anthropic:claude-3-5-haiku", tools=mcp_tools)
        await disconnect_mcp()
    else:
        yield create_agent("anthropic:claude-3-5-haiku", tools=[])

Simple Factory Without Context

from langgraph_sdk.runtime import ServerRuntime

async def my_factory(runtime: ServerRuntime) -> CompiledGraph:
    user = runtime.ensure_user()
    return build_graph(user)

Checking Access Context

from langgraph_sdk.runtime import ServerRuntime

async def my_factory(runtime: ServerRuntime):
    if runtime.access_context == "threads.create_run":
        print("Graph will be fully executed")
    elif runtime.access_context == "assistants.read":
        print("Graph structure being introspected")
    return build_graph()

Using the Store

from langgraph_sdk.runtime import ServerRuntime

async def my_factory(runtime: ServerRuntime):
    # Store is always available regardless of access context
    store = runtime.store
    config = await store.get(("config",), "default")
    return build_graph(config=config)

Related Pages

Page Connections

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