Runner

The Runner encapsulates framework-specific execution strategies, providing a consistent interface for running agents across different frameworks. You can skip this section if you are not planning to contribute to Agent Kernel.

Overview

What is a Runner?

A Runner:

• Executes framework-specific agent logic (run())
• Streams AK stream events (text/reasoning deltas, tool calls) for frameworks that support it (stream())
• Converts Agent Kernel request models to framework-native input, and framework output back to AgentReply models
• Manages framework session state within the Agent Kernel Session
• Creates the ToolContext so tools can access the runtime, agent, session, and requests

Runner Interface

from abc import ABC, abstractmethod
from typing import AsyncGenerator
from agentkernel.core import Session
from agentkernel.core.event import StreamEvent
from agentkernel.core.model import AgentReply, AgentRequest

class Runner(ABC):
    @abstractmethod
    async def run(self, agent: "Agent", session: Session, requests: list[AgentRequest]) -> AgentReply:
        """Execute the agent with the given requests within the session context."""

    @property
    def supports_streaming(self) -> bool:
        """Whether this runner streams; adapters without streaming support override to False."""
        return True

    @abstractmethod
    async def stream(self, agent: "Agent", session: Session, requests: list[AgentRequest]) -> AsyncGenerator[StreamEvent, None]:
        """Yield AK stream events for streaming execution (execution.mode: stream)."""

• run() takes a list of typed requests (AgentRequestText, AgentRequestImage, AgentRequestFile, AgentRequestAny), not a raw prompt string, and returns an AgentReply.
• stream() is an async generator of StreamEvent members (core/event.py: MessageStart/TextDelta/MessageEnd, ReasoningStart/ReasoningDelta/ReasoningEnd, ToolCallStart/ToolCallArgs/ToolCallEnd/ToolCallResult, StepStart/StepEnd) — never a bare str. Runtime.stream() wraps each event in a StreamChunk (delta is populated only for TextDelta) and passes it through post-hook filtering before it reaches the client. See docs/specs/523-ag-ui-support/spec.md for the full event-mapping rules.

Framework Runners

Runner	Framework	AK Runner.stream()
OpenAIRunner	OpenAI Agents SDK	✅ (Runner.run_streamed)
LangGraphRunner	LangGraph	✅ (astream_events)
GoogleADKRunner	Google ADK	✅ (SSE streaming mode)
PydanticAIRunner	Pydantic AI	✅ (run_stream_events)
MAFRunner	Microsoft Agent Framework	✅ (agent.run(..., stream=True))
CrewAIRunner	CrewAI	❌ raises NotImplementedError
SmolagentsRunner	Smolagents	❌ raises NotImplementedError

Each runner follows the same shape internally:

class OpenAIRunner(Runner):
    async def run(self, agent, session, requests):
        # 1. Restore framework-specific session state from the AK session
        # 2. Convert AgentRequest models to framework-native input
        # 3. Create ToolContext, execute the framework's run API
        # 4. Save updated framework state back into the session
        # 5. Convert the result to AgentReplyText / AgentReplyImage / AgentReplyAny

Reply Types

Every runner returns an AgentReply from run(). The union covers three reply models:

Type	Produced when	Payload
AgentReplyText	The agent produces plain text (default)	text: str
AgentReplyImage	The agent produces text plus an image	text: str, image_data: str
AgentReplyAny	The agent is configured for structured output	content: dict

All reply types carry the prompt that was sent to the agent.

Structured replies: AgentReplyAny

When an agent is configured to produce structured output (see the per-framework "Structured Output" sections under Frameworks), the runner detects it and returns an AgentReplyAny instead of coercing the result to a string:

from agentkernel.core.model import AgentReplyAny

reply = await runner.run(agent, session, requests)
if isinstance(reply, AgentReplyAny):
    data = reply.content          # dict, no re-parsing needed

• content holds the structured result as a JSON-compatible dict. Pydantic model results are converted with model_dump(mode="json").
• str(reply) returns the JSON-serialized content, so any consumer that renders replies as text (chat integrations, logging, tracing) works unchanged.
• Plain-text agents are unaffected and continue to return AgentReplyText.

Streaming and structured output

Structured output applies to non-streaming execution only — run() returns AgentReplyAny directly. Streamed runs emit StreamEvents (text/reasoning/tool-call deltas) rather than a parsed reply; a streaming client reconstructs the final structured result from the tool-call events for frameworks (like Pydantic AI) that stream the structured answer as a tool call. This is unrelated to per-framework truncation behavior — see each framework's page for streaming fidelity notes.

Streaming Execution

When execution.mode: stream is configured, the pipeline calls Runner.stream() instead of run():

async for event in runner.stream(agent, session, requests):
    print(event)   # a StreamEvent, e.g. TextDelta, ToolCallStart, MessageEnd

In practice you rarely call this directly; use AgentService.stream_multi() or the REST API, which wrap each event in StreamChunk objects (delta, event, done, error, session_id) and run every event through the post-hook on_stream_event() chain:

async for chunk in service.stream_multi(requests):
    if chunk.error:
        ...
    elif chunk.delta:
        print(chunk.delta, end="")   # populated only for TextDelta events

delta is populated only when chunk.event is a TextDelta, so a plain-text consumer can keep concatenating delta unchanged; a consumer that wants tool calls, reasoning, or message/step boundaries reads the full event. Frameworks without native token streaming (CrewAI, Smolagents) declare supports_streaming = False and raise NotImplementedError from stream(); use the default synchronous mode (or rest_sync on AWS) with those frameworks.

Execution Flow

Note that hooks, session locking, and persistence are handled by Runtime.run() around the runner; the runner itself only deals with framework execution and state conversion. See Execution Flow for the full pipeline.

Using Runners

Runners are typically accessed through agents, and invoked via the Runtime (which applies hooks and persistence):

from agentkernel.core import Runtime
from agentkernel.core.model import AgentRequestText

runtime = Runtime.current()
agent = runtime.agents().get("assistant")
session = runtime.sessions().get("user-123") or runtime.sessions().new("user-123")

# Preferred: run through the Runtime so hooks and persistence apply
reply = await runtime.run(agent, session, [AgentRequestText(prompt="Hello")])

For most applications, the higher-level AgentService is more convenient than touching runners at all.

Session Integration

Runners work closely with Sessions to maintain state. Each framework stores its own state under its own key:

async def run(self, agent, session, requests):
    # Get framework-specific state from the AK session
    framework_state = session.get("openai")   # e.g. "openai", "langgraph", ...

    if not framework_state:
        framework_state = self._create_state()

    result = await self._execute(agent, framework_state, requests)

    session.set("openai", framework_state)    # persisted by Runtime after the run
    return result

From inside a hook or a tool (i.e. while the agent is executing), read the currently-running agent's framework state without naming its runner key explicitly via session.get_framework_session().

Per-run framework context

In addition to their own internal state, runners honour one reserved session value, the framework context — a framework-agnostic, per-run context/state dict carried across turns. Seed and read it with session.set_framework_context() / get_framework_context() from a pre- or post-hook (see Session → Framework context / per-run state). When a context is set, a runner:

1. Loads a deep copy of the stored context before invoking the framework.
2. Injects it into the native framework call (mapped to each framework's own context/state mechanism).
3. Writes back the produced state — shallow-merged over the loaded copy (framework-touched top-level keys win, untouched caller keys are preserved) — but only after the native call succeeds, so a crashed or disconnected run leaves the previously stored context intact.

How faithfully a caller dict round-trips is not uniform across frameworks:

Framework	Fidelity	Injected as	Written back
OpenAI	Full round-trip	Runner.run(..., context=ctx) — tools mutate it in place	the same object, in full
Pydantic AI	Full round-trip	agent.run(..., deps=ctx) — native tools mutate it in place via RunContext.deps (caveats)	the same object, in full
Microsoft Agent Framework	Full round-trip	native_session.state["ak_context"] — tools mutate it via FunctionInvocationContext.session.state	the context dictionary, including tool-added keys
Google ADK	Round-trips (filtered), accumulate-only	merged into the ADK session state (AK-internal keys always win, so they cannot be displaced by a caller key)	the accumulated session state, minus AK-internal and app:/user:/temp:-prefixed keys — tool-added keys survive, and so does anything else written to the state (caveats)
Smolagents	Round-trips (filtered)	agent.run(..., additional_args=ctx) — which smolagents also appends to the task prompt (caveat)	agent.state restricted to pre-seeded keys — brand-new keys are dropped
LangGraph	Declared channels only	spread into the graph input alongside messages (written last, so a caller key cannot replace it)	only keys the graph's state schema declares as channels (prebuilt agents drop unknown keys)
CrewAI	Unsupported	not injected	none — a set context is ignored, with one warning logged per runner

Because of this divergence, tool authors who want a context write to be portable across every framework should pre-seed every key they intend to write before the run.

Per-agent native run options

Each framework's run API takes options of its own: a turn cap, lifecycle hooks, a run config. Agent Kernel does not abstract over them. Instead, an application declares them per agent through the module, in the framework's own types, and the runner merges them into its native call:

from agents import RunConfig

OpenAIModule([agent]).run_options(
    agent,
    max_turns=25,
    hooks=ProgressHooks(),                                   # an SDK RunHooks subclass
    run_config=RunConfig(call_model_input_filter=trim),      # any RunConfig field
)

Module.run_options(agent, [factory], **options) is chained like pre_hook / post_hook; repeated calls merge, the later call winning per key. The declared dict lives on Agent.run_options, is read on every run, and is never persisted.

The merge rule. The runner copies the declared options and writes the keys it owns last, so a declared option can never displace a value the adapter populates (the session, the framework context, the input). The same keys are also rejected at declaration: each adapter names them in RESERVED_RUN_OPTIONS, and run_options raises ValueError naming the key and the reason. Unknown keys are not validated and reach the SDK. OpenAI, ADK, Pydantic AI, Microsoft Agent Framework and smolagents raise TypeError for an unknown keyword on the first run; LangGraph and CrewAI drop one silently (ainvoke ignores unknown keywords, and Crew is a Pydantic model that ignores extra fields), so check the spelling there. The one guarded case is LangGraph: a RunnableConfig key declared at the top level (recursion_limit, callbacks, ...) is rejected with a message pointing to config={...}.

Two adapters share an object with the caller rather than a key, and merge deeper:

• LangGraph config is deep-merged with the config the runner builds: dict-valued keys merge with the runner's entries winning (configurable.thread_id stays the session id), list-valued keys (callbacks, tags) concatenate with the runner's entries first (a tracing runner's handler is kept), everything else is the caller's.
• ADK run_config in stream mode is copied with streaming_mode=SSE, because the stream mapping depends on partial events; one warning is logged per runner when the caller explicitly set a different streaming_mode.

Framework	Options go to	Reserved keys	Turn limit	Progress hook
OpenAI	Runner.run / run_streamed	starting_agent, input, session, context, conversation_id, previous_response_id, auto_previous_response_id (the SDK rejects the last three alongside the session the runner passes)	max_turns	hooks=RunHooks()
LangGraph	ainvoke / astream_events	input, version, stream_mode, output_keys, print_mode, config.configurable.thread_id, and any RunnableConfig key at the top level	config["recursion_limit"]	config["callbacks"]
Google ADK	the per-run App (plugins), the per-run Runner(...) constructor (memory_service, artifact_service, credential_service, plugin_close_timeout) and run_async (run_config)	agent, app, app_name, node, session_service, auto_create_session, user_id, session_id, new_message, state_delta, invocation_id, yield_user_message	RunConfig(max_llm_calls=...)	plugins=[BasePlugin()]
Pydantic AI	agent.run / run_stream_events	user_prompt, message_history, deps	UsageLimits(request_limit=...)	event_stream_handler (run mode; dropped with one warning in stream mode)
Microsoft Agent Framework	agent.run (including stream=True) — model settings are nested in options	messages, session, stream	native function-invocation configuration on the client	native middleware on the agent or client
CrewAI	the per-run Crew(...) constructor (verbose=False is an overridable default; max_rpm is a forwarded rate limit)	agents, tasks, memory	max_iter on the native Agent (needs nothing from Agent Kernel)	step_callback / task_callback
Smolagents	agent.run	task, reset, additional_args, stream, return_full_result	max_steps	step_callbacks on the agent constructor (needs nothing from Agent Kernel)

Computed per run. Module.run_options(agent, [factory], **options) also takes one optional positional argument after the agent, a RunOptionsFactory: (agent, session, requests) -> mapping, sync or async. The runner resolves the options once per run through Agent.resolve_run_options(session, requests): a copy of the static dict, then the factory's result validated against RESERVED_RUN_OPTIONS and merged over it at the top level (a factory key wins; a dict-valued option it returns replaces the static one wholesale, so on LangGraph extend the static config from agent.run_options rather than returning a bare one), then the keys Agent Kernel owns written last as above. An agent without a factory resolves to exactly its static dict, so every existing declaration is unchanged.

def options_for(agent, session, requests):
    options = {"run_config": RunConfig(trace_metadata={"session_id": session.id})}
    if session.id.startswith("guest"):
        options["max_turns"] = 10  # over the static 25 below
    return options

OpenAIModule([agent]).run_options(agent, options_for, max_turns=25, hooks=ProgressHooks())

The factory runs after the pre-hooks, so it sees the request list the run will use, and inside the run, so Session.current(), Agent.current() and ToolContext.get() resolve in it; on Google ADK the tool context is created later in the run, so ToolContext.get() raises RuntimeError there and the factory's own arguments carry the same information. One factory per agent (a later call replaces it; keywords keep merging), one object shared by every concurrent run, so keep per-run state in the session. A factory that raises, returns a non-mapping or names a reserved key fails that run the way a framework error does: a user-facing error reply in run, a propagated exception in stream. Nothing is cached, so the failure repeats until the factory is fixed.

Progress: two paths. In execution.mode: stream, PostHook.on_stream_event sees Agent Kernel's framework-agnostic StreamEvents for whatever the adapter maps. Native hooks through run options work in any mode, including rest_sync, and expose the framework's full lifecycle (LLM start, agent start, handoffs). A native hook's callbacks run inside the Agent Kernel run, so Session.current() and Agent.current() resolve in them; a hook instance declared once at load time therefore needs no per-request plumbing. It is shared by every concurrent run of that agent, so keep per-run state in the session (its volatile cache is cleared after each run), never on the hook.

Run options compose with tracing: the Langfuse, Logfire, OpenLLMetry and CloudWatch runners delegate to the base runner, so a declared hooks= still reaches the SDK when trace.enabled is on. The custom-runner path remains for behaviour that is not a keyword argument of the native call.

See the per-framework demos: examples/cli/openai-run-options, langgraph-run-options, adk-run-options, pydanticai-run-options, maf-run-options, crewai-run-options and smolagents-run-options, and examples/cli/openai-dynamic-run-options for options computed per run.

Best Practices

Async Execution

Always use await when calling runners:

# Correct
reply = await runtime.run(agent, session, requests)

# Incorrect
reply = runtime.run(agent, session, requests)  # Returns coroutine

Error Handling

Wrap execution in try-except:

try:
    reply = await runtime.run(agent, session, requests)
except Exception as e:
    logger.error(f"Runner error: {e}")
    # Handle error appropriately

Summary

• Runners execute framework-specific agent logic and expose both run() and stream()
• Each framework has its own Runner implementation
• OpenAI Agents SDK, LangGraph, Google ADK, Pydantic AI, and Microsoft Agent Framework support native token streaming (as AK StreamEvents); the CrewAI and Smolagents adapters do not implement it yet
• Runners convert typed requests/replies and manage framework session state
• Runners inject the reserved framework_context into the native call and write the produced state back on success (fidelity varies per framework)
• Always use async/await, and prefer Runtime.run()/AgentService over calling runners directly

Next Steps

• Session Management
• Module Organization
• Framework Integration
• Execution Flow