codekingpro/portable-devtools

1from __future__ import annotations2 3import sys4from dataclasses import dataclass, field5from typing import TYPE_CHECKING, Generic, Literal, TypeVar6 7if sys.version_info >= (3, 13):8    ContextT = TypeVar("ContextT", default=None)9else:10    ContextT = TypeVar("ContextT")11 12if sys.version_info >= (3, 12):13    from typing import TypeAliasType14else:15    from typing_extensions import TypeAliasType16 17from langgraph_sdk.auth.types import BaseUser18 19if TYPE_CHECKING:20    from langgraph.store.base import BaseStore21 22__all__ = [23    "AccessContext",24    "ServerRuntime",25]26 27 28AccessContext = Literal[29    "threads.create_run",30    "threads.update",31    "threads.read",32    "assistants.read",33]34 35 36@dataclass(kw_only=True, slots=True, frozen=True)37class _ServerRuntimeBase(Generic[ContextT]):38    """Base for server runtime variants.39 40    !!! warning "Beta"41        This API is in beta and may change in future releases.42    """43 44    access_context: AccessContext45    """Why the graph factory is being called.46 47    The server accesses graphs in several contexts beyond just executing runs.48    For example, it calls the graph factory to retrieve schemas, render the49    graph structure, or read state history. This field tells you which50    operation triggered the current call.51 52    In all contexts, the returned graph must have the same topology (nodes,53    edges, state schema) as the graph used for execution. Use54    `.execution_runtime` to conditionally set up expensive *resources*55    (MCP servers, DB connections) without changing the graph structure.56 57    Write contexts (graph is used to write state):58 59    - `threads.create_run` (`graph.astream`) — full graph execution60      (nodes + edges). `context` is available (use `.execution_runtime`61      to narrow).62    - `threads.update` (`graph.aupdate_state`) — does NOT execute node63      functions or evaluate edges. Only runs the node's channel writers64      to apply the provided values to state channels as if the specified65      node had returned them. Reducers are applied and channel triggers66      are set, so the next `invoke`/`stream` call will evaluate edges67      from that node to determine the next step. Does not need access to68      external resources, but a different graph topology will apply69      writes to the wrong channels.70 71    Read state contexts (graph used to format the returned72    `StateSnapshot`). A different topology may cause `get_state` to73    report incorrect pending tasks. Note that `useStream` uses the state74    history endpoint to render interrupts and support branching:75 76    - `threads.read` (`graph.aget_state`, `graph.aget_state_history`) —77      the graph structure informs which tasks to include in the prepared78      view of the latest checkpoint and how to process subgraphs.79 80    Introspection contexts (graph structure only, no execution).81    A different topology may cause schemas and visualizations to not82    match actual execution:83 84    - `assistants.read` (`graph.aget_graph`, `graph.aget_subgraphs`,85      `graph.aget_schemas`) — return the graph definition, subgraph86      definitions, and input/output/config schemas. Used for87      visualization in the studio UI and to populate schemas for MCP,88      A2A, and other protocol integrations.89    """90 91    user: BaseUser | None = field(default=None)92    """The authenticated user, or `None` if no custom auth is configured."""93 94    store: BaseStore95    """Store for the graph run, enabling persistence and memory."""96 97    @property98    def execution_runtime(self) -> _ExecutionRuntime[ContextT] | None:99        """Narrow to the execution runtime, or `None` if not in an execution context.100 101        When the server calls the graph factory for `threads.create_run`, the returned102        object provides access to `context` (typed by the graph's103        `context_schema`). For all other access contexts (introspection, state104        reads, state updates), this returns `None`.105 106        Use this to conditionally set up expensive resources (MCP tool servers,107        database connections, etc.) that are only needed during execution:108 109        ```python110        import contextlib111        from langgraph_sdk.runtime import ServerRuntime112 113        @contextlib.asynccontextmanager114        async def my_factory(runtime: ServerRuntime[MyCtx]):115            if ert := runtime.execution_runtime:116                # Only connect to MCP servers when actually executing a run.117                # Introspection calls (get_schema, get_graph, ...) skip this.118                mcp_tools = await connect_mcp(ert.context.mcp_endpoint)119                yield create_agent(model, tools=mcp_tools)120                await disconnect_mcp()121            else:122                yield create_agent(model, tools=[])123        ```124        """125        if isinstance(self, _ExecutionRuntime):126            return self127        return None128 129    def ensure_user(self) -> BaseUser:130        """Return the authenticated user, or raise if not available.131 132        When custom auth is configured, `user` is set for all access contexts133        (the factory is only called from HTTP handlers where the auth134        middleware has already run). This method raises only when no custom135        auth is configured.136 137        Raises:138            PermissionError: If no user is authenticated.139        """140        if self.user is None:141            raise PermissionError(142                f"No authenticated user available in access_context='{self.access_context}'. "143                "Ensure custom auth is configured for the server."144            )145        return self.user146 147 148@dataclass(kw_only=True, slots=True, frozen=True)149class _ExecutionRuntime(_ServerRuntimeBase[ContextT], Generic[ContextT]):150    """Runtime for `threads.create_run` — the graph will be fully executed.151 152    Access this via `.execution_runtime` on `ServerRuntime`. Do not153    construct directly.154 155    !!! warning "Beta"156        This API is in beta and may change in future releases.157    """158 159    context: ContextT = field(default=None)  # type: ignore[assignment]160    """The graph run context, typed by the graph's `context_schema`.161 162    Only available during `threads.create_run`.163    """164 165 166@dataclass(kw_only=True, slots=True, frozen=True)167class _ReadRuntime(_ServerRuntimeBase[ContextT], Generic[ContextT]):168    """Runtime for non-execution access contexts.169 170    Used for introspection (`assistants.read`), state operations171    (`threads.read`), and state updates (`threads.update`).172    No `context` is available.173 174    !!! warning "Beta"175        This API is in beta and may change in future releases.176    """177 178 179ServerRuntime = TypeAliasType(180    "ServerRuntime",181    _ExecutionRuntime[ContextT] | _ReadRuntime[ContextT],182    type_params=(ContextT,),183)184"""Runtime context passed to graph builder factories within the Agent Server.185 186Requires version 0.7.30 or later of the agent server.187 188The server calls your graph factory in multiple contexts: executing runs,189reading state, fetching schemas, and more. `ServerRuntime` provides190the authenticated user, store, and access context for every call. Use191`.execution_runtime` to narrow to the execution variant and access192`context`.193 194Example — conditionally initialize MCP tools only during execution:195 196```python197import contextlib198from dataclasses import dataclass199 200from langchain.agents import create_agent201from langgraph_sdk.runtime import ServerRuntime202from my_agent import connect_mcp, disconnect_mcp203 204@dataclass205class MyCtx:206    mcp_endpoint: str207 208_readonly_agent = create_agent("anthropic:claude-3-5-haiku", tools=[])209 210@contextlib.asynccontextmanager211async def my_factory(runtime: ServerRuntime[MyCtx]):212    if ert := runtime.execution_runtime:213        # Only connect to MCP servers for actual runs.214        # Schema / graph introspection calls skip this.215        user_id = runtime.ensure_user().identity216        mcp_tools = await connect_mcp(ert.context.mcp_endpoint, user_id)217        yield create_agent("anthropic:claude-3-5-haiku", tools=mcp_tools)218        await disconnect_mcp()219    else:220        yield _readonly_agent221```222 223Example — simple factory that ignores context:224 225```python226from langgraph_sdk.runtime import ServerRuntime227 228def build_graph(user: BaseUser) -> CompiledGraph:229    ...230 231async def my_factory(runtime: ServerRuntime) -> CompiledGraph:232    # No generic needed if you don't use context.233    return build_graph(runtime.ensure_user())234```235 236!!! warning "Beta"237    This API is in beta and may change in future releases.238"""239