Skip to main content
This page describes the concrete AionRuntimeContext surface used by Aion LangGraph executions. It is the context schema passed to LangGraph and is populated by Aion Server for each invocation. AionRuntimeContext is the lower-level request context. Thread and Message are authoring helpers derived from that context and are injected automatically by create_event_router when a handler declares them.

​
Overview

Register AionRuntimeContext as the graph context schema: 
from langgraph.graph import StateGraph

from aion.core.runtime import AionRuntimeContext


builder = StateGraph(State, context_schema=AionRuntimeContext)
Access it from LangGraph nodes through Runtime[AionRuntimeContext]: 
from langgraph.runtime import Runtime

from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring.runtime import Thread


async def node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
    context = runtime.context
    thread = Thread.from_context(context)
    event = context.event

    if event is not None:
        await thread.reply(f"Received event: {event.kind.value}")

    return {}

​
Properties

PropertyPurpose
inboxRaw A2AInbox snapshot for hybrid authoring and low-level debugging.
eventTyped inbound Aion event, or None for direct A2A requests without an Aion event envelope.
distribution_extension_payloadParsed Distribution extension payload when the request includes one.
graph_kwargsExtra framework-specific values passed through from the graph runtime.
The runtime context stores the Distribution extension payload in its protocol shape. It does not project distribution, behavior, or environment fields onto an identity object.

​
Distribution Accessors

Use these methods when a request includes a Distribution extension payload:
MethodReturns
get_distribution()Distribution model for the current invocation, or None.
get_behavior()Behavior model for the invoked agent implementation, or None.
get_environment()Environment model for the runtime configuration, or None.
get_principal_identity()Principal identity associated with the distribution, or None.
get_service_identity()Service identity associated with the distribution, or None.
get_principal_identity() and get_service_identity() are intentionally separate helpers. The principal identity is the Aion principal that owns the distribution when one exists. The service identity represents an external network identity associated with the distribution when one exists.

​
Event Handlers

When using create_event_router, handlers may declare higher-level injected parameters instead of reaching through runtime.context manually: 
async def handle_message(thread, message, distribution, principal_identity):
    await thread.reply(f"Hello {message.user.id}")
These injected values are convenience projections built from AionRuntimeContext. They are not stored as top-level fields on the runtime context model.

​
Why This Lives in Runtime Context

LangGraph gives us a clean separation between:
  • graph state, which is ideal for LLM-facing data such as state.messages
  • invocation-scoped runtime context, which is ideal for immutable request data and dependency injection
That is a better fit for Aion than storing routing metadata in graph state, because:
  • the inbound messaging target is part of the current request, not long-term graph memory
  • history lookups and outbound side effects should not require custom state reducers
  • protocol payloads can remain available without leaking into model-facing graph state

​
LangGraph thread_id vs Aion Thread IDs

LangGraph already uses config["configurable"]["thread_id"] for checkpointing and persistence. That should remain LangGraph’s own persistence identifier. Aion messaging thread or context identifiers are derived from the inbound event payload and exposed through the Thread helper. Those IDs can be related to LangGraph checkpoint IDs, but they should not be conflated.

​
Example

from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime

from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring.runtime import Thread


async def router(state: MessagesState, runtime: Runtime[AionRuntimeContext]) -> dict:
    context = runtime.context
    thread = Thread.from_context(context)
    event = context.event

    if event is not None and event.kind.value == "to.aion.distribution.command.1.0.0":
        await thread.reply("Command received.")
        return {}

    return {}


builder = StateGraph(MessagesState, context_schema=AionRuntimeContext)
builder.add_node("router", router)
builder.add_edge(START, "router")
builder.add_edge("router", END)
graph = builder.compile()