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 token deltas 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.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."""

    @abstractmethod
    async def stream(self, agent: "Agent", session: Session, requests: list[AgentRequest]) -> AsyncGenerator[str, None]:
        """Yield token deltas 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 raw token strings. Runtime.stream() wraps each delta in a StreamChunk and passes it through post-hook filtering before it reaches the client.

Framework Runners

Runner	Framework	Native token streaming
OpenAIRunner	OpenAI Agents SDK	✅ (Runner.run_streamed)
LangGraphRunner	LangGraph	✅ (astream_events)
GoogleADKRunner	Google ADK	✅ (SSE streaming mode)
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 limitation

Structured output applies to non-streaming execution only. Streamed runs emit token-by-token text deltas and are not parsed into structured replies.

Streaming Execution

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

async for delta in runner.stream(agent, session, requests):
    print(delta, end="")   # raw token strings

In practice you rarely call this directly; use AgentService.stream_multi() or the REST API, which wrap the deltas in StreamChunk objects (delta, done, error, session_id) and run the post-hook on_stream_chunk() filter on every token:

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

Frameworks without native token streaming (CrewAI, Smolagents) raise NotImplementedError; 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

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, and Google ADK support token streaming; CrewAI and Smolagents do not
• Runners convert typed requests/replies and manage framework session state
• Always use async/await, and prefer Runtime.run()/AgentService over calling runners directly

Next Steps

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