---
name: ak-dev-new-framework-integration
description: >
  Step-by-step guide for adding a new agent framework adapter to Agent Kernel.
  Use this skill when you need to integrate a new agent framework (beyond OpenAI,
    CrewAI, LangGraph, Google ADK, Smolagents, Pydantic AI). Covers creating the adapter module, implementing
  Agent/Runner/Module subclasses, adding optional dependencies, exports, and tests.
license: Apache-2.0
metadata:
  author: yaalalabs
  category: developer
---

# Adding a New Framework Integration

This guide walks through adding support for a new agent framework to Agent Kernel. Use the existing OpenAI adapter (`ak-py/src/agentkernel/framework/openai/`) as the canonical reference implementation.

## Prerequisites

- Understand the architecture skill (`.agents/skills/ak-dev-architecture/SKILL.md`)
- Familiarity with the target framework's API
- The target framework must support async execution (or provide an async wrapper)

## Step-by-Step

### 1. Create the Framework Adapter Directory

```
ak-py/src/agentkernel/framework/<name>/
├── __init__.py
└── <name>.py
```

Replace `<name>` with the framework's lowercase identifier (e.g., `openai`, `langgraph`).

### 2. Implement the Session State Class (if needed)

If the framework requires per-session state (e.g., conversation history), create a session data class:

```python
class <Name>Session:
    """Stores framework-specific session data."""
    def __init__(self):
        self._history = []  # or whatever state the framework needs

    def get_history(self):
        return self._history

    def add_to_history(self, item):
        self._history.append(item)

    def clear_session(self):
        self._history.clear()
```

The session data is stored in the Agent Kernel `Session` via `session.set("<name>", <Name>Session())` and retrieved via `session.get("<name>")`. This key **must** be the same string passed as your Runner's `name` (see Step 3) — hook authors reach it via `Session.get_framework_session()`, which resolves `Agent.current().runner.name` under the hood.

### 3. Implement the Runner

Subclass `Runner` from `agentkernel.core.base`:

```python
from agentkernel.core.base import Runner, Session
from agentkernel.core.model import AgentReply, AgentReplyText, AgentRequest, AgentRequestText
from agentkernel.core.tool import ToolContext

FRAMEWORK = "<name>"

class <Name>Runner(Runner):
    def __init__(self):
        # must match the session key below — Session.get_framework_session() resolves it
        # via Agent.current().runner.name
        super().__init__(FRAMEWORK)

    def _session(self, session: Session) -> <Name>Session:
        """Get or create framework-specific session data."""
        data = session.get(FRAMEWORK)
        if data is None:
            data = <Name>Session()
            session.set(FRAMEWORK, data)
        return data

    async def run(self, agent, session: Session, requests: list[AgentRequest]) -> AgentReply:
        # 1. Create ToolContext for tool functions to access
        tool_context = ToolContext(
            runtime=Runtime.current(),
            agent=agent,
            session=session,
            requests=requests
        )

        with tool_context:
            tool_context.set()
            try:
                # 2. Get framework-specific session state
                fw_session = self._session(session)

                # 3. Convert AgentRequest models to framework-native format
                # e.g., extract text from AgentRequestText
                prompt = ""
                for req in requests:
                    if isinstance(req, AgentRequestText):
                        prompt = req.prompt

                # 4. Call the framework's execution API
                result = await self._execute(agent, fw_session, prompt)  # framework-specific

                # 5. Update session state
                fw_session.add_to_history({"input": prompt, "output": result})

                # 6. Return as AgentReply
                return AgentReplyText(response=str(result), prompt=prompt)
            finally:
                tool_context.reset()
```

**Key requirements:**
- Always create a `ToolContext` and set it so tool functions can access `ToolContext.get()`
- Always reset `ToolContext` in a `finally` block
- Handle all `AgentRequest` subtypes (`AgentRequestText`, `AgentRequestImage`, `AgentRequestFile`)
- Return an `AgentReply` (`AgentReplyText` or `AgentReplyImage`)
- Resolve the run options once per `run()` and once per `stream()` with `options = await agent.resolve_run_options(session, requests)` (#758), after the request-shape early returns and before the framework-context load, then build the native call's keyword arguments with `self._native_kwargs(options, <ak_owned>=...)` (#754), never a fixed keyword set: the resolved per-agent run options (the static declaration with a per-run factory's result merged over it) are copied first and the keys your adapter populates (the session, the framework context, the input) are written last. Pass the same mapping to any helper of yours that reads options; never read `agent.run_options` at a call site. If the framework's options object and yours are one object (LangGraph's `config`) or must be adjusted per mode (ADK's `RunConfig` in stream mode), merge or copy it before passing it as an AK-owned key, and do the same in `stream()`

### 3b. Implement `stream()` with AK Stream Events

`Runner` declares `stream()` as `@abstractmethod`, returning `AsyncGenerator[StreamEvent, None]` —
**every** adapter must implement it, even if the framework doesn't support token streaming, and it
must yield `StreamEvent` members (`core/event.py`: `MessageStart`/`TextDelta`/`MessageEnd`,
`ReasoningStart`/`ReasoningDelta`/`ReasoningEnd`, `ToolCallStart`/`ToolCallArgs`/`ToolCallEnd`/
`ToolCallResult`, `StepStart`/`StepEnd`) — never a bare `str`. A runner that yields a bare `str` is
rejected by `StreamChunk.event` with a `pydantic.ValidationError`; there is no string-normalisation
fallback in `Runtime.stream()`.

**If the framework's SDK exposes a token-delta stream**, map its native events onto AK events —
bracket assistant text with `MessageStart`/`MessageEnd` (deferred until text actually arrives, so a
tool-only turn doesn't emit an empty message), and map tool-call/tool-result events onto
`ToolCallStart`/`ToolCallArgs`/`ToolCallEnd`/`ToolCallResult` correlated by the framework's own call
id where one exists (never generate an id when the framework supplies one — a generated id cannot
correlate a result to the call that produced it):

```python
from collections.abc import AsyncGenerator

from ...core.event import MessageEnd, MessageStart, StreamEvent, TextDelta

async def stream(self, agent, session: Session, requests: list[AgentRequest]) -> AsyncGenerator[StreamEvent, None]:
    tool_context = ToolContext(Runtime.current(), agent, session, requests)
    try:
        tool_context.set()
        fw_session = self._session(session)
        prompt = "".join(req.prompt for req in requests if isinstance(req, AgentRequestText))

        # Anything remembered mid-stream is a local — see the rule below.
        message_id: str | None = None

        result = await self._execute_streamed(agent, fw_session, prompt)  # framework-specific
        message_id: str | None = None  # local — the runner is shared across sessions
        async for event in result:
            delta = self._extract_text_delta(event)  # framework-specific
            if delta:
                if message_id is None:
                    message_id = uuid4().hex
                    yield MessageStart(message_id=message_id)
                yield TextDelta(message_id=message_id, content=delta)
        if message_id is not None:
            yield MessageEnd(message_id=message_id)
    finally:
        tool_context.reset()
```

**If the framework has no native token streaming** (e.g. CrewAI, smolagents), override `supports_streaming` to `False` and implement `stream()` as a generator that always raises, so a caller can check the property before invoking `stream()` instead of provoking the raise, while `stream()` itself still satisfies the abstract method contract and fails fast with a clear message:

```python
async def stream(self, agent: Any, session: Session, requests: list[AgentRequest]) -> AsyncGenerator[StreamEvent, None]:
    """
    :return: False — this adapter does not implement streaming, so stream() always raises.
    """
    return False

async def stream(self, agent: Any, session: Session, requests: list[AgentRequest]) -> AsyncGenerator[StreamEvent, None]:
    """
    <Name> streaming is not implemented in this adapter yet.
    :raises NotImplementedError: Always raised — use rest_sync mode instead.
    """
    raise NotImplementedError(
        "<Name> streaming is not implemented in the Agent Kernel adapter yet. Use rest_sync mode."
    )
    yield  # make this an async generator to satisfy the type contract

@property
def supports_streaming(self) -> bool:
    """Declared False so a caller can reject a streamed request instead of provoking the raise."""
    return False
```

`Runtime.stream()` runs every yielded event through `PostHook.on_stream_event()` (#670), wraps what
survives in a `StreamChunk` (`delta` is populated only for `TextDelta`, so a plain-text consumer that
only reads `StreamChunk.delta` keeps working unchanged), and forwards it to the caller (REST SSE
endpoint or AWS Lambda WebSocket/SQS pipeline). No other core changes are needed to support a new
framework's streaming — just implement `Runner.stream()`. See `docs/specs/523-ag-ui-support/spec.md`
for the full event-mapping rules and per-adapter correlation-id/boundary-derivation decisions, and
`docs/specs/670-streaming-post-hooks/` for the hook contract your events pass through.

Two consequences for a new adapter, both from #670: a hook may now drop or rewrite **any** event you
emit, including boundaries, so do not assume what you yield is what the client receives; and a hook
raising `StreamHalt` abandons your generator mid-iteration, so anything your `stream()` does after the
loop (writing back framework context, for instance) will not run on a halted run.

### 3c. Wire up the per-run framework context

The base `Runner` provides two helpers so a caller-supplied, framework-agnostic context/state dict
(seeded by a hook via `session.set_framework_context(...)`) rides across turns. **Your `run()` and
`stream()` must call them and map the one AK-level dict onto your framework's native
context/state mechanism** (or decline it explicitly, as CrewAI does):

- `incoming = self._load_framework_context(session)` — call **before** the native invocation.
  Returns a **deep copy** of the stored dict, or `None` when the key is absent. When `None`, inject
  nothing (framework default) — this keeps the no-context path unchanged for existing apps.
- Inject `incoming` (when not `None`) via the framework's native mechanism (a run `context=`, an
  input state channel, a session-state delta, `additional_args=`, …).
- After a **successful** native call, extract the framework's post-run state as `produced` and call
  `self._store_framework_context(session, incoming, produced)`. This shallow-merges `produced` over
  `incoming` (framework-touched top-level keys win; untouched caller keys preserved) and fail-fast
  checks picklability before writing back.

```python
# In run(), inside the existing try, around the native call:
incoming = self._load_framework_context(session)
result = await self._execute(agent, fw_session, prompt, context=incoming)  # inject natively
produced = self._extract_state(result, incoming)  # framework-specific; may be a subset of keys
self._store_framework_context(session, incoming, produced)  # only after a successful call
```

**Placement matters (atomicity):** put the write-back **inside the `try`, after the native call,
before the `except`** — and for `stream()`, **after the `async for` loop but still inside the
`try`**, never in `finally`. A framework error or a client disconnect (`GeneratorExit`) then unwinds
before it, leaving the previously stored context intact rather than persisting partial state.

**Seed AK-internal keys last.** If you inject the caller's dict by merging it into a native state
dict that also carries AK-internal entries, assign the internal ones **after** the caller's keys so a
caller key can never displace them (`ak_tool_context` in ADK, `messages` in LangGraph). The failures
this prevents are silent and confusing — a broken tool-context lookup, or a replaced message list.

**Watch for injection side effects.** A framework's "context" slot is not always private: smolagents'
`additional_args` is merged into the agent state *and* appended to the task prompt, so the caller's
dict reaches the model. If your framework does something similar, document it on the framework's page
so callers know not to put secrets in `framework_context`.

**Declare your round-trip fidelity honestly** in the framework's docs and the fidelity table in
`docs/docs/core-concepts/runner.md` — how much of a caller dict actually survives depends on the
framework (full round-trip, filtered to seeded keys, declared-channels-only, or unsupported). Name the
**native handle a tool uses** to reach the context (`RunContextWrapper.context` on OpenAI,
`RunContext.deps` on Pydantic AI, `tool_context.state` on ADK, …) — tools use that, never the `Session`
accessors. If the framework has no safe caller-state slot, do **not** inject; instead log a single
warning per runner instance and skip both load and write-back (see the CrewAI adapter for the pattern).

### 4. Implement the Agent Wrapper

Subclass `Agent` from `agentkernel.core.base`:

```python
from agentkernel.core.base import Agent, Runner, Session

class <Name>Agent(Agent):
    def __init__(self, name: str, runner: Runner, native_agent):
        super().__init__(name, runner)
        self._native_agent = native_agent

    def get_description(self) -> str:
        # Return the agent's description from the native framework object
        return self._native_agent.instructions  # framework-specific

    def get_a2a_card(self):
        from agentkernel.core.builder import A2ACardBuilder
        return A2ACardBuilder.build(
            name=self.name,
            description=self.get_description(),
            skills=[...]  # extract from agent tools
        )
```

**Key requirements:**
- `get_description()` must return a meaningful description from the native framework agent
- `get_a2a_card()` must return a valid A2A agent card built via `A2ACardBuilder`
- Store the native agent in `self._native_agent` for access in the Runner
- Declare `RESERVED_RUN_OPTIONS: ClassVar[Mapping[str, str]]` (#754): every keyword your runner passes itself, plus any that changes the result shape your reply mapping reads, each mapped to a one-line reason. `Module.run_options` rejects them at declaration with that reason. Override `validate_run_options` only for a nested key (LangGraph's `config.configurable.thread_id`)

### 5. Implement the ToolBuilder

Subclass `ToolBuilder` from `agentkernel.core.tool`:

```python
from agentkernel.core.tool import ToolBuilder

class <Name>ToolBuilder(ToolBuilder):
    @classmethod
    def bind(cls, funcs: list) -> list:
        """Wrap plain Python functions into framework-native tool objects."""
        tools = []
        for func in funcs:
            # Convert func to framework-specific tool format
            tool = framework_specific_tool_wrapper(func)
            tools.append(tool)
        return tools
```

Add `**options` and forward them when the framework's own tool wrapper takes per-tool arguments that
change what the tool *is*. Three adapters do — OpenAI (`function_tool`), Google ADK (`FunctionTool`)
and Pydantic AI (`Tool`) — and in each the framework's human-in-the-loop gate (`needs_approval`,
`require_confirmation`, `requires_approval`) is exactly such an argument: without forwarding, a gated
tool could only be declared natively, giving up the framework-free function the builder exists to
allow. Options apply to every function in the call, so tools wanting different ones are bound in
separate calls. Do not add the parameter when the framework has no such arguments — smolagents'
`tool()` takes only the function.

### 6. Implement the Module

Subclass `Module` from `agentkernel.core.module`:

```python
from agentkernel.core.module import Module
from agentkernel.trace.trace import Trace

class <Name>Module(Module):
    def __init__(self, agents: list):
        super().__init__()
        # Check if tracing is enabled, use traced runner if so
        trace_runner = Trace.get().<name>()  # returns Runner or None
        self.runner = trace_runner if trace_runner else <Name>Runner()
        self.load(agents)

    def _wrap(self, agent, agents) -> <Name>Agent:
        return <Name>Agent(agent.name, self.runner, agent)

    def load(self, agents: list) -> "Module":
        return super().load(agents)

    # pre_hook / post_hook / run_options are inherited from Module: they resolve the wrapped agent
    # through _native_agent_name(agent), which defaults to agent.name. Override that hook only when
    # your framework registers agents under something else (CrewAI: role).
```

**Key requirements:**
- Constructor takes native framework agents, creates a Runner, calls `self.load(agents)`
- `_wrap()` creates the Agent wrapper — the agent `name` must come from the native agent
- Support trace runners via `Trace.get().<name>()`
- `pre_hook`, `post_hook` and `run_options` are inherited and concrete (they share `Module._wrapped`, which raises `ValueError` for an agent not loaded in the module); override `_native_agent_name(agent)` only when the native agent is not registered under `agent.name` (CrewAI uses `role`, smolagents a fallback name)

### 7. Create the `__init__.py`

```python
# ak-py/src/agentkernel/framework/<name>/__init__.py
from .<name> import <Name>Module, <Name>ToolBuilder
```

### 8. Create the Public API Alias

Create `ak-py/src/agentkernel/<name>.py`:

```python
from .framework.<name> import <Name>Module, <Name>ToolBuilder
```

This allows users to import as `from agentkernel.<name> import <Name>Module`.

### 9. Update Package Exports

Add the framework to `ak-py/src/agentkernel/__init__.py` if appropriate (following the existing pattern).

### 10. Add Optional Dependencies

In `ak-py/pyproject.toml`, add an optional dependency group:

```toml
[project.optional-dependencies]
<name> = [
    "framework-package>=x.y.z",
    # Add any instrumentation packages for tracing support
]
```

### 11. Add Tracing Support

There are **four** built-in tracing backends, each with per-framework traced runners. A new framework needs a traced runner under **each** of `ak-py/src/agentkernel/trace/langfuse/<name>.py`, `trace/openllmetry/<name>.py`, `trace/logfire/<name>.py`, and `trace/cloudwatch/<name>.py` (the CloudWatch runner takes the `CloudWatch` tracer and wraps `super().run()` in `self._tracer.span(...)`):

```python
from ...framework.<name>.<name> import <Name>Runner

class LangFuse<Name>Runner(<Name>Runner):
    def __init__(self, langfuse_client):
        super().__init__()
        self._client = langfuse_client

    async def run(self, agent, session, requests):
        with self._client.start_as_current_span(name=agent.name):
            return await super().run(agent, session, requests)
```

Also add a new abstract framework method in `ak-py/src/agentkernel/trace/base.py` and the corresponding `Trace.<name>()` method in `ak-py/src/agentkernel/trace/trace.py`.

### 12. Add Tests

Create tests in `ak-py/tests/`:

```python
# ak-py/tests/test_<name>_runner.py
# ak-py/tests/test_tool_<name>.py
```

Follow the existing test patterns (e.g. `test_openai_runner.py`, `test_smolagents_runner.py`, `test_pydanticai_runner.py`, `test_tool_adk.py`) — use `DummyRunner`, `DummyAgent` for unit tests, `monkeypatch` for config overrides, `@pytest.mark.asyncio` for async tests.

### 13. Add Examples

Create at minimum:
- `examples/cli/<name>/` — CLI demo with `demo.py`, `pyproject.toml`, `demo_test.py`
- `examples/api/<name>/` — API demo (optional but recommended)

### 14. Add Documentation

- Add a page under `docs/docs/frameworks/<name>.md` — note the page slug may differ from the adapter directory name (e.g. the `adk` adapter's page is `docs/docs/frameworks/google-adk.md`, referenced as `'frameworks/google-adk'` in `docs/sidebars.js`)
- Update `docs/sidebars.js` to include the new framework
- Landing page (`docs/src/components/*/data.tsx`): add a tile to the **Agent frameworks & dev tools** row in `IntegrationsMarquee/data.tsx` (role `Framework`, `href` to the new page, logo or `react-icons/si` glyph); add `pick("<tile name>")` to the **Agent frameworks** card in `ArchitectureOverview/data.tsx`; add the framework to the **Framework Adapters** card's `tags` and `description` under the Build tab in `FeatureExplorer/data.tsx`. Logo sourcing and the build check are in `ak-dev-sync-docs-from-branch`, *Docs-Site Landing and Features Pages*
- Features page (`docs/src/pages/features.tsx`): the `integrations` list and the "Framework adapters for N SDKs" highlight on the Six Core Abstractions card. Grep `docs/src/pages/*.tsx` and `docs/docs/intro.md` for the framework roll call and add the new name wherever the others are listed

## Checklist

- [ ] `ak-py/src/agentkernel/framework/<name>/` directory with `__init__.py` and `<name>.py`
- [ ] `<Name>Session` (if needed), `<Name>Runner`, `<Name>Agent`, `<Name>Module`, `<Name>ToolBuilder`
- [ ] `<Name>Runner.stream()` implemented — either real event streaming or a `NotImplementedError` stub
- [ ] `<Name>Runner.supports_streaming` declared — `False` when `stream()` only raises, so callers reject instead of provoking it
- [ ] `<Name>Runner.supports_pause` declared (#606) — leave it at the inherited `False` unless the framework has a **durable, programmatic** pause. Unlike `supports_streaming` it defaults to False, so a framework without one needs no declaration and `resume()`/`resume_stream()` keep their raising defaults
- [ ] When it does pause: detection placed **before** the existing reply mapping (a paused run still fills the answer field with the text produced on the way to stopping), the record written with `PausedRunState.add` inside the `try` after a successful call, and any framework-specific rejection raised **above** the `try` — the adapter's `except Exception` would otherwise flatten it into a generic reply
- [ ] When it does pause: **you** clear the record with `PausedRunState.clear` on the success path, inside the `try`. `Runtime` deliberately does not tidy up after you — it cannot tell a failed resume from an ordinary reply, so a fallback there would destroy the pause on a transient error. A record still present after a resume means that resume failed and the decision can be sent again
- [ ] When it does pause and your framework holds **one conversation per session**, replace with `PausedRunState.clear_for_runner(session, self.name)` rather than clearing the list — another framework's agent may hold an independent, still-answerable pause on the same session. Only OpenAI appends, because its `RunState` is a self-contained snapshot
- [ ] When it does pause: `resume()` and `resume_stream()` produce the **same shapes `run()` and `stream()` do** — the structured-reply mapping, the event envelope, the reasoning events. Three of ADK's review findings were a resumed turn quietly differing from an ordinary one; share a helper between the pair rather than writing the second copy
- [ ] `<Name>Runner`'s `name` (passed to `super().__init__()`) matches the session key used in `session.get/set(...)` — required for `Session.get_framework_session()` to resolve it
- [ ] Public alias at `ak-py/src/agentkernel/<name>.py`
- [ ] Optional dependency group in `ak-py/pyproject.toml`
- [ ] Trace runners in `ak-py/src/agentkernel/trace/{langfuse,openllmetry,logfire,cloudwatch}/<name>.py` (optional)
- [ ] Updates to `trace/base.py` and `trace/trace.py` (if adding tracing)
- [ ] `<Name>Agent.RESERVED_RUN_OPTIONS` declared; `run()` and `stream()` each resolve once with `await agent.resolve_run_options(session, requests)` after the early returns and pass the mapping to `_native_kwargs` and your helpers (#754, #758)
- [ ] `_native_agent_name` overridden if the registered name is not `agent.name`
- [ ] Unit tests in `ak-py/tests/`, including: declared run options reach the native call, an AK-owned key wins over a bypassed declaration, the declared dict is not mutated across runs, and the reserved keys are rejected through the module; a factory-merged mapping (a mock whose `resolve_run_options` returns it) wins over the static key at the native call, and `resolve_run_options` is awaited exactly once per `run` and per `stream` (#758)
- [ ] CLI example in `examples/cli/<name>/`
- [ ] Documentation in `docs/docs/frameworks/<name>.md`
- [ ] Landing page inventories: marquee tile (`IntegrationsMarquee/data.tsx`), `pick()` chip on the Agent frameworks card (`ArchitectureOverview/data.tsx`), Framework Adapters card tags (`FeatureExplorer/data.tsx`)
- [ ] Framework inventories on `docs/src/pages/features.tsx` (`integrations` and SDK count) and in `docs/docs/intro.md`