ak-dev-new-framework-integration

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:

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:

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()