Integration Patterns

This page describes the three authoring styles LangGraph developers can choose from when running behind Aion. The goal is not to force one SDK shape onto every graph. Aion should work well for plain LangGraph graphs, for explicit A2A-aware graphs, and for a higher-level fluent SDK surface.

1. Plain LangGraph

Use plain LangGraph when you only need normal conversational input and output.

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, AnyMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


def reply_node(state: State) -> dict:
    reply = model.invoke(state["messages"])
    return {"messages": [reply]}


builder = StateGraph(State)
builder.add_node("reply", reply_node)
builder.add_edge(START, "reply")
builder.add_edge("reply", END)
graph = builder.compile()

In this mode:

Aion maps inbound text into state.messages
Aion infers the final reply from streamed model output or the last agent-authored AIMessage
the graph does not need any Aion-specific dependency

This is the best fit for basic chat agents.

2. Hybrid A2A

Use hybrid authoring when you want raw protocol access without adopting the higher-level authoring helpers.

from typing import Annotated, Optional, TypedDict

from a2a.types import Message, Part, Role
from aion.core.a2a import A2AOutbox
from aion.core.runtime import AionRuntimeContext
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.runtime import Runtime


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    a2a_outbox: Optional[A2AOutbox]


def reply_node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
    inbound = runtime.context.inbox.message
    return {
        "a2a_outbox": A2AOutbox(
            message=Message(
                context_id=inbound.context_id,
                role=Role.ROLE_AGENT,
                parts=[Part(text="Structured reply")],
            )
        )
    }


builder = StateGraph(State, context_schema=AionRuntimeContext)

In this mode:

runtime.context.inbox gives you direct access to the inbound task, message, and request metadata
a2a_outbox gives you direct control over the outbound A2A object
LangGraph still provides ordinary state.messages for model-facing logic

This is the best fit when you need structured parts, explicit metadata, or fine-grained protocol control.

3. SDK-aware LangGraph

Use the aion.langgraph.authoring surface when you want normalized messaging, runtime-scoped Aion context, model-service helpers, and MCP tool loading without hand-assembling A2A envelopes.

from typing import Annotated, TypedDict

from langchain_core.messages import AnyMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.runtime import Runtime

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


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


async def reply_node(state: State, runtime: Runtime[AionRuntimeContext]) -> dict:
    thread = Thread.from_context(runtime.context)
    await thread.reply("I received that.")
    return {}


builder = StateGraph(State, context_schema=AionRuntimeContext)
builder.add_node("reply", reply_node)
builder.add_edge(START, "reply")
builder.add_edge("reply", END)
graph = builder.compile()

In this mode:

request-scoped routing data stays in LangGraph runtime context, not graph state
thread.reply(...), thread.post(...), and thread.typing(...) emit through LangGraph custom streaming events
model helpers inject Aion model-service authentication and runtime principal attribution per request
MCP helpers resolve static and runtime capability references after AionRuntimeContext is available

This is the best fit when you want a higher-level authoring experience while still mapping to Aion’s generic A2A extensions.

Event router

For event-driven integrations, prefer create_event_router. It creates a normal LangGraph node and injects only the parameters your handler declares:

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

from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring import Message, Thread, create_event_router


async def handle_message(thread: Thread, message: Message) -> None:
    await thread.reply(f"Got your message: {message.text}")


builder = StateGraph(MessagesState, context_schema=AionRuntimeContext)
builder.add_node("aion_events", create_event_router(on_message=handle_message))
builder.add_edge(START, "aion_events")
builder.add_edge("aion_events", END)
graph = builder.compile()

Handlers may declare any subset of state, runtime, context, event, distribution, behavior, environment, principal_identity, service_identity, inbox, thread, or message. Other declared parameters are left for LangGraph-native injection.

Model service

Use aion_chat_model or aion_chat_openai when LangChain model calls should go through Aion’s OpenAI-compatible model service:

from aion.langgraph.authoring import aion_chat_model

model = aion_chat_model("model-id-from-control-plane", temperature=0.2)

The helper owns Aion connection settings such as api_key, base_url, default_headers, http_client, and http_async_client. Pass model behavior options such as temperature, token limits, timeouts, and retries.

MCP tools

Use load_aion_mcp_tools after an AionRuntimeContext exists. Static capability references address known control-plane endpoints, while runtime references resolve their subject from the incoming request:

from aion.api import (
    CapabilityReference,
    CapabilitySubjectSource,
    RuntimeCapabilityReference,
)
from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring import load_aion_mcp_tools


async def load_tools(context: AionRuntimeContext):
    return await load_aion_mcp_tools(
        context,
        capability_references=[
            CapabilityReference.global_mcp(),
        ],
        runtime_capability_references=[
            RuntimeCapabilityReference.primary_mcp(
                CapabilitySubjectSource.INCOMING_DISTRIBUTION
            ),
        ],
    )

Use AionLangGraphMcpResolver when you want to reuse the same MCP resolution settings across invocations.

Which Pattern to Choose

Pattern	Best for	Main tradeoff
Plain LangGraph	Simple text-first chat flows	Least control over protocol details
Hybrid A2A	Full A2A control with minimal new abstractions	More verbose authoring
SDK-aware LangGraph	Thread, message, event-router, MCP, and streaming helpers	Adds an authoring dependency

It is important that these modes compose cleanly. A graph should be able to use the fluent SDK for one turn, a2a_outbox for another, and plain LangGraph fallback for the rest without switching adapters.

Python SDK

LangGraph

Google ADK

Integration Patterns

1. Plain LangGraph

2. Hybrid A2A

3. SDK-aware LangGraph

Event router

Model service

MCP tools

Which Pattern to Choose

​1. Plain LangGraph

​2. Hybrid A2A

​3. SDK-aware LangGraph

​Event router

​Model service

​MCP tools

​Which Pattern to Choose

1. Plain LangGraph

2. Hybrid A2A

3. SDK-aware LangGraph

Event router

Model service

MCP tools

Which Pattern to Choose