Wrap a framework for chat

The Agent Adapter chat page talks to one function: agent(message, session), returning text or an async iterator of text chunks (Chat with your agent). Each wrapper below turns a framework's own entry point into that function, in a few lines.

flowchart LR
  P["chat page"] --> F["agent(message, session)"]
  F --> LG["LangGraph app.ainvoke"]
  F --> ADK["ADK Runner.run_async"]
  F --> CR["CrewAI crew.kickoff_async"]
  F --> H["your HTTP backend"]

Tool calls show as cards when they run through session.executor (with ctx=session.context()); a framework that runs its tools itself shows only its answer. A framework's own "approved" never counts: only the page's one-use decision does.

LangGraph

A compiled graph with a messages state. Earlier turns go in as (role, text) pairs.

def langgraph_agent(app: Any) -> ChatAgent:
    """A compiled LangGraph app (messages state) as a chat agent."""

    async def reply(message: ChatMessage, session: ChatSession) -> str:
        history = [(m.role, m.content) for m in session.history]
        out = await app.ainvoke({"messages": [*history, ("user", message.text)]})
        return str(out["messages"][-1].content)

    return reply

ADK

Each chat is one ADK session, made on first use; ADK keeps the history itself. An edit or Replay branch has a new session.id, so ADK starts a fresh session there without the earlier turns (seed it from session.history if the agent needs them).

def adk_agent(runner: Any, user_id: str = "local") -> ChatAgent:
    """An ADK `Runner` as a chat agent; each chat is one ADK session."""

    async def reply(message: ChatMessage, session: ChatSession) -> AsyncIterator[str]:
        from google.genai import types  # installed with google-adk

        sessions, app = runner.session_service, runner.app_name
        if await sessions.get_session(app_name=app, user_id=user_id, session_id=session.id) is None:
            await sessions.create_session(app_name=app, user_id=user_id, session_id=session.id)
        content = types.Content(role="user", parts=[types.Part(text=message.text)])
        async for event in runner.run_async(
            user_id=user_id, session_id=session.id, new_message=content
        ):
            if event.is_final_response() and event.content and event.content.parts:
                yield "".join(part.text or "" for part in event.content.parts)

    return reply

CrewAI

The crew's task reads the user's text as {question}.

def crewai_agent(crew: Any) -> ChatAgent:
    """A CrewAI `Crew` as a chat agent; its task reads `{question}`."""

    async def reply(message: ChatMessage, session: ChatSession) -> str:
        result = await crew.kickoff_async(inputs={"question": message.text})
        return str(result.raw)

    return reply

Your HTTP backend

A FastAPI (or any) service answering POST {message, session_id} with {answer}. The function is sync, so it runs in a worker thread.

def http_agent(url: str, timeout: float = 60.0) -> ChatAgent:
    """Your own backend (FastAPI or any) that answers POST {message, session_id} with {answer}."""

    def reply(message: ChatMessage, session: ChatSession) -> str:  # sync: runs in a thread
        body = json.dumps({"message": message.text, "session_id": session.id}).encode("utf-8")
        headers = {"Content-Type": "application/json"}
        request = urllib.request.Request(url, data=body, headers=headers)
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return str(json.loads(response.read())["answer"])

    return reply

Run one

Pass the wrapped agent to serve, for example serve(agent=langgraph_agent(app)), or put it in a module and start agent-adapter ui --agent app:chat. The wrappers are tested with stand-ins for each framework; LangGraph also runs for real when it is installed.