Integrate MCP tools into ADK and LangGraph agents

Agent frameworks turn MCP tools into tools their model can call, but each one wraps the protocol differently: what the model sees, how an SDK connection is held, and what happens when a tool fails. We give the typed task-tracker server from Implement an MCP server with typed tool contracts to two agents, a Google ADK agent through McpToolset and a LangGraph agent through langchain-mcp-adapters. Both run Gemma gemma4:e2b locally through Ollama. You inspect what each framework discovers, run a successful task, make a tool fail, and test both agent loops without a model.

When you need this. You already run a tool server, like the task tracker, and two teams want their agents to use it: one builds on Google ADK, the other on LangGraph. Nobody should write the tools twice. Both agents can discover the same server over MCP; what differs is how each framework hands the tools to its model and what the model sees when a call fails.

Asked to close “the docs task”, the ADK agent lists the tasks to find the ID, then completes it:

Here ClientSession names an SDK connection context and its initialization handshake. It is not an ADK conversation session, which retains messages and state. A stdio server is a separate process; creating another SDK connection may start another process. The protocol and transport version determine whether an HTTP session ID exists.

1. Set up the agent environment

Download the complete lab ZIP, which includes the Python files, fixtures, tests and README. Extract it before running the commands below. The folder includes a copy of tasks_server.py, identical to the first lesson’s file. Install the agent packages from requirements-lock.txt in a second virtual environment. Download the typed-tools lab beside this ZIP: its SDK environment starts the server, while the agent environment contains the framework adapters. Keep the extracted directories adjacent:

unzip -n implement-an-mcp-server-with-typed-tool-contracts.zip
python3.12 -m venv mcp-typed-tools/.venv
mcp-typed-tools/.venv/bin/python -m pip install -r mcp-typed-tools/requirements-lock.txt
unzip integrate-mcp-tools-into-adk-and-langgraph-agents.zip
cd mcp-agent-frameworks
python3.12 -m venv .venv-agents
.venv-agents/bin/python -m pip install -r requirements-lock.txt
export TASKS_SERVER_PYTHON="$(cd ../mcp-typed-tools && pwd)/.venv/bin/python"
export OLLAMA_API_BASE=http://localhost:11434

Here the two environments are required. langchain-mcp-adapters 0.3.2 requires an MCP SDK older than 2.0, so the agent side runs mcp 1.30.0, while the server uses mcp 2.3.0. That works because MCP is a wire protocol between two processes: the two SDKs agree on a protocol version during connection initialization, and neither imports the other. The same split appears in production whenever an agent and a tool server are released by different teams.

Pull the model with ollama pull gemma4:e2b. The recorded runs used digest b37049369adf. Gemma 4 is released under the Apache 2.0 license.

2. Give the tools to an ADK agent

Now we hand the server to the first framework. McpToolset is ADK’s MCP client. Given StdioConnectionParams, it starts the server as a child process, lists its tools when the agent first needs them, and converts each one into an ADK tool. ADK reaches Ollama through LiteLLM, using the ollama_chat/ model prefix.

"""A Google ADK agent that uses the tasks MCP server.

Run: python agent_adk.py --tools                  (list what the toolset discovered)
     python agent_adk.py "Which tasks are open?"  (needs Ollama with gemma4:e2b)
TASKS_SERVER_PYTHON is the Python that has mcp 2.3.0 installed for tasks_server.py.
"""
import asyncio
import os
import sys
from pathlib import Path

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.adk.tools.mcp_tool import McpToolset, StdioConnectionParams
from google.genai import types
from mcp import StdioServerParameters

tasks_server = StdioServerParameters(
    command=os.environ.get("TASKS_SERVER_PYTHON", sys.executable),
    args=[str(Path(__file__).with_name("tasks_server.py"))],
)
toolset = McpToolset(connection_params=StdioConnectionParams(server_params=tasks_server))

root_agent = Agent(
    name="tasks_agent",
    model=LiteLlm(model="ollama_chat/gemma4:e2b"),
    instruction="You manage a software project's task tracker. Use the tools to read and change it.",
    tools=[toolset],
)


async def discover() -> None:
    """Print each tool as ADK declares it to the model."""
    for tool in await toolset.get_tools():
        schema = tool._get_declaration().parameters_json_schema
        print("tool  ", tool.name, schema.get("required", []), list(schema["properties"]))
    await toolset.close()


async def ask(question: str, agent: Agent = root_agent) -> list:
    """Run one question through the agent and return every event."""
    runner = InMemoryRunner(agent=agent, app_name="tasks")
    session = await runner.session_service.create_session(app_name="tasks", user_id="dev")
    message = types.Content(role="user", parts=[types.Part(text=question)])
    events = [event async for event in runner.run_async(user_id="dev", session_id=session.id, new_message=message)]
    await runner.close()
    return events


def show(events: list) -> None:
    for event in events:
        for part in event.content.parts if event.content else []:
            if part.function_call:
                print("call  ", part.function_call.name, part.function_call.args)
            elif part.function_response:
                response = part.function_response.response
                print("result", "error" if response.get("isError") else "ok", response.get("structuredContent") or response["content"][0]["text"])
            elif part.text and not part.thought:
                print("answer", part.text.strip())


if __name__ == "__main__":
    if sys.argv[1] == "--tools":
        asyncio.run(discover())
    else:
        show(asyncio.run(ask(sys.argv[1])))

Use McpToolset. ADK 2.12.0 keeps MCPToolset only as a subclass that warns that it is deprecated. InMemoryRunner keeps the conversation in memory; each ask() call creates a new session, and runner.close() also closes the toolset’s SDK connection and stdio child process.

complete_task changes data, and a model chooses when to call it. McpToolset accepts require_confirmation, a boolean or a function, to ask a person before a tool runs, and tool_filter to expose only some tools, such as the read-only list_tasks. Treat the server’s read_only_hint as a hint for that choice, not as a permission check.

Unknown task ID in ADK. McpToolset returns the MCP result with isError: true as the function response. The agent continues and the model reads the tool’s correction; keep useful ToolError text on the server.

3. Give the same tools to a LangGraph agent

MultiServerMCPClient holds the connection settings. create_agent from LangChain builds a LangGraph graph that alternates between the model and a tool node until the model answers in text.

"""A LangGraph agent that uses the tasks MCP server.

Run: python agent_langgraph.py --tools                  (list what the adapter loaded)
     python agent_langgraph.py "Which tasks are open?"  (needs Ollama with gemma4:e2b)
TASKS_SERVER_PYTHON is the Python that has mcp 2.3.0 installed for tasks_server.py.
"""
import asyncio
import os
import sys
from pathlib import Path

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain_ollama import ChatOllama

client = MultiServerMCPClient({
    "tasks": {
        "transport": "stdio",
        "command": os.environ.get("TASKS_SERVER_PYTHON", sys.executable),
        "args": [str(Path(__file__).with_name("tasks_server.py"))],
    }
})


async def discover() -> None:
    """Print each tool as the adapter gives it to the model."""
    async with client.session("tasks") as session:
        for tool in await load_mcp_tools(session):
            print("tool  ", tool.name, tool.args_schema.get("required", []), list(tool.args_schema["properties"]))


async def ask(question: str, model=None) -> list:
    """Run one question through the agent in one MCP session and return every message."""
    async with client.session("tasks") as session:
        agent = create_agent(
            model or ChatOllama(model="gemma4:e2b"),
            await load_mcp_tools(session),
            system_prompt="You manage a software project's task tracker. Use the tools to read and change it.",
        )
        result = await agent.ainvoke({"messages": [{"role": "user", "content": question}]})
    return result["messages"]


def show(messages: list) -> None:
    for message in messages:
        for call in getattr(message, "tool_calls", []):
            print("call  ", call["name"], call["args"])
        if message.type == "tool":
            print("result", message.status, message.artifact.get("structured_content") if message.artifact else message.text)
        elif message.type == "ai" and message.text:
            print("answer", message.text.strip())


if __name__ == "__main__":
    if sys.argv[1] == "--tools":
        asyncio.run(discover())
    else:
        show(asyncio.run(ask(sys.argv[1])))

The important decision is how long the SDK connection and server process stay open. client.get_tools() is shorter, but each tool call it returns opens a new SDK client connection and stdio server process. ask() opens one client.session("tasks") and loads the tools from it, so every call in one question reaches the same server process and sees the same state. The state-loss test below, which contrasts get_tools() with one client.session() shows the difference.

With this stdio adapter, each newly opened SDK connection starts a Python server process, so ask() starts one server process per question and get_tools() starts one per tool call. For a busy agent, move the server to Streamable HTTP so that a new client connection does not require a new server process.

Unknown task ID in LangGraph. The adapter returns a ToolMessage with status="error" and continues the run. The model can read the reason and choose a valid ID.

handle_tool_errors=False. With this argument, load_mcp_tools raises ToolException instead of returning an error message. Keep the default handling if the model should recover; otherwise install an application handler before expecting the run to continue.

State lost between tool calls. get_tools() opens a new SDK connection and stdio process for each call, so a newly added task is gone on the next call. Load tools from one client.session() when calls must share the server’s in-memory state.

4. Check what each framework discovered

Here is what each framework will give its model. Before you run a model, check that discovery worked. Each script’s --tools flag lists the tools as the framework will give them to the model, with each tool’s required and total parameters:

$ python agent_adk.py --tools
tool   add_task ['title'] ['title', 'priority']
tool   complete_task ['task_id'] ['task_id']
tool   list_tasks [] ['status']

$ python agent_langgraph.py --tools
tool   list_tasks [] ['status']
tool   add_task ['title'] ['title', 'priority']
tool   complete_task ['task_id'] ['task_id']

Both frameworks pass the server’s JSON Schema through unchanged, including the pattern on task_id. ADK lists the tools in alphabetical order; LangGraph keeps the server’s order. This is the complete_task declaration ADK sends to the model:

{
  "description": "Mark one task as done and return it.",
  "name": "complete_task",
  "parameters_json_schema": {
    "properties": {
      "task_id": {
        "description": "A task ID such as T-1, as returned by list_tasks.",
        "pattern": "^T-\\d+$",
        "title": "Task Id",
        "type": "string"
      }
    },
    "required": [
      "task_id"
    ],
    "type": "object",
    "title": "complete_taskArguments"
  },
  "response_json_schema": {
    "properties": {
      "id": {
        "description": "Task ID such as T-1.",
        "title": "Id",
        "type": "string"
      },
      "title": {
        "title": "Title",
        "type": "string"
      },
      "priority": {
        "enum": [
          "low",
          "normal",
          "high"
        ],
        "title": "Priority",
        "type": "string"
      },
      "status": {
        "enum": [
          "open",
          "done"
        ],
        "title": "Status",
        "type": "string"
      }
    },
    "required": [
      "id",
      "title",
      "priority",
      "status"
    ],
    "type": "object",
    "description": "One task in the tracker.",
    "title": "Task"
  }
}

Transport failures, such as a wrong TASKS_SERVER_PYTHON, are not tool errors. The adapters raise them regardless of handle_tool_errors, so the run stops before the model sees anything. Run the server command by hand first when an agent finds no tools.

5. Run both agents on Gemma

.venv-agents/bin/python agent_adk.py "Close the docs task."
.venv-agents/bin/python agent_langgraph.py "Close the docs task."
.venv-agents/bin/python agent_adk.py "Mark task T-9 as done."
.venv-agents/bin/python agent_langgraph.py "Mark task T-9 as done."

$ python agent_adk.py "Close the docs task."
call   list_tasks {}
result ok {'tasks': [{'id': 'T-1', 'title': 'Write the release notes', 'priority': 'normal', 'status': 'open'}, {'id': 'T-2', 'title': 'Fix the login redirect bug', 'priority': 'high', 'status': 'done'}, {'id': 'T-3', 'title': 'Update the API docs', 'priority': 'low', 'status': 'open'}], 'count': 3}
call   complete_task {'task_id': 'T-3'}
result ok {'id': 'T-3', 'title': 'Update the API docs', 'priority': 'low', 'status': 'done'}
answer Task 'Update the API docs' (T-3) has been successfully closed.

$ python agent_langgraph.py "Close the docs task."
call   list_tasks {}
result success {'tasks': [{'id': 'T-1', 'title': 'Write the release notes', 'priority': 'normal', 'status': 'open'}, {'id': 'T-2', 'title': 'Fix the login redirect bug', 'priority': 'high', 'status': 'done'}, {'id': 'T-3', 'title': 'Update the API docs', 'priority': 'low', 'status': 'open'}], 'count': 3}
call   complete_task {'task_id': 'T-3'}
result success {'id': 'T-3', 'title': 'Update the API docs', 'priority': 'low', 'status': 'done'}
answer Task "Update the API docs" (T-3) has been closed.

$ python agent_adk.py "Mark task T-9 as done."
call   complete_task {'task_id': 'T-9'}
result error Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.
answer I could not find a task with the ID T-9. Please use the `list_tasks` tool to see the current list of available task IDs.

$ python agent_langgraph.py "Mark task T-9 as done."
call   complete_task {'task_id': 'T-9'}
result error Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.
answer I was unable to mark task T-9 as done because no task with that ID was found. Please list the tasks to see the valid IDs.

In both frameworks, Gemma resolved “the docs task” to T-3 by calling list_tasks first, because only list_tasks returns IDs and the complete_task description says so. For T-9, both agents passed the ID through, got a tool error and told the user.

record_runs.py ran each question 3 times per framework with Ollama’s default options and kept every trial in agent-runs.json. Every trial reached the same outcome: T-3 closed, or the T-9 error reported. In 1 of the 3 ADK trials for T-9, Gemma followed the error message’s advice and called list_tasks before it answered; LangGraph did that in 0. The runs took between 10.12 and 31.65 seconds each. This is a small sample, not a reliability measurement; a test that asserts an exact call sequence from a live model will be flaky, which is why the next step replays recorded turns instead.

StdioConnectionParams.timeout, 5 seconds by default, also configures the SDK connection’s request-response wait. A slow tool can therefore raise an MCP timeout after the connection is established. This is not an overall agent or model-call deadline, and a timeout does not establish that the server cancelled the operation. Model latency dominated these recorded runs; set an application deadline that covers both model requests and tool calls.

6. Test the agent loop without a model

A test that calls Ollama is slow and not repeatable. test_agents.py replaces only the model: a replay model returns the recorded model turns in order, while the real server, MCP clients and agent loops run. If a framework changes how it calls tools or formats results, the transcript no longer matches the recording and the test fails.

.venv-agents/bin/python -W ignore -m unittest -v test_agents.py

test_adk_agent_resolves_a_task_name_to_its_id (test_agents.AgentReplay.test_adk_agent_resolves_a_task_name_to_its_id) ... Skipping missing token usage metadata for agent tasks_agent and model replay
Skipping missing token usage metadata for agent tasks_agent and model replay
Skipping missing token usage metadata for agent tasks_agent and model replay
ok
test_adk_returns_a_tool_error_to_the_model (test_agents.AgentReplay.test_adk_returns_a_tool_error_to_the_model) ... Tool 'complete_task' failed: 'Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.'
Skipping missing token usage metadata for agent tasks_agent and model replay
Skipping missing token usage metadata for agent tasks_agent and model replay
ok
test_agent_server_is_the_typed_tools_lesson_server (test_agents.AgentReplay.test_agent_server_is_the_typed_tools_lesson_server) ... ok
test_both_frameworks_discover_the_recorded_tools (test_agents.AgentReplay.test_both_frameworks_discover_the_recorded_tools) ... ok
test_every_recorded_trial_reached_the_expected_outcome (test_agents.AgentReplay.test_every_recorded_trial_reached_the_expected_outcome) ... ok
test_get_tools_starts_a_new_server_session_for_each_call (test_agents.AgentReplay.test_get_tools_starts_a_new_server_session_for_each_call) ... ok
test_langgraph_agent_resolves_a_task_name_to_its_id (test_agents.AgentReplay.test_langgraph_agent_resolves_a_task_name_to_its_id) ... ok
test_langgraph_returns_a_tool_error_to_the_model (test_agents.AgentReplay.test_langgraph_returns_a_tool_error_to_the_model) ... Tool 'complete_task' failed: 'Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.'
ok
test_langgraph_without_error_handling_raises_instead (test_agents.AgentReplay.test_langgraph_without_error_handling_raises_instead) ... Tool 'complete_task' failed: 'Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.'
ok
test_one_session_keeps_server_state_between_calls (test_agents.AgentReplay.test_one_session_keeps_server_state_between_calls) ... ok

----------------------------------------------------------------------
Ran 10 tests in 5.567s

OK

All 10 tests pass. -W ignore hides ADK’s experimental-feature warnings so the output stays readable.

"""Replay the recorded model turns through both agents, offline, against the real MCP server.

Run: python -m unittest -v test_agents.py
The model is replaced by its recorded turns; the MCP server, adapters and agent loops are real.
TASKS_SERVER_PYTHON (or AIML_MCP_PYTHON) must point at a Python with mcp 2.3.0 installed.
"""
import asyncio
import json
import os
import unittest
from pathlib import Path

os.environ.setdefault("TASKS_SERVER_PYTHON", os.environ.get("AIML_MCP_PYTHON", ""))

from google.adk.models.base_llm import BaseLlm  # noqa: E402
from google.adk.models.llm_response import LlmResponse  # noqa: E402
from google.genai import types  # noqa: E402
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel  # noqa: E402
from langchain_core.messages import AIMessage  # noqa: E402
from langchain_core.tools import ToolException  # noqa: E402
from langchain_mcp_adapters.tools import load_mcp_tools  # noqa: E402

import agent_adk  # noqa: E402
import agent_langgraph  # noqa: E402
import record_runs  # noqa: E402

HERE = Path(__file__).resolve().parent
RECORD = json.loads((HERE / "agent-runs.json").read_text(encoding="utf-8"))


def recorded(framework, question):
    """The first recorded trial, which the replay tests reproduce."""
    run = next(run for run in RECORD["runs"] if run["framework"] == framework and run["question"] == question)
    return run["trials"][0]


class ReplayLlm(BaseLlm):
    """An ADK model that answers with the recorded turns, in order."""
    turns: list

    async def generate_content_async(self, llm_request, stream=False):
        yield LlmResponse(content=types.Content.model_validate(self.turns.pop(0)))


class ReplayChatModel(GenericFakeChatModel):
    """A LangChain chat model that answers with the recorded turns, in order."""

    def bind_tools(self, tools, **kwargs):
        return self


def replay_adk(question):
    agent = agent_adk.root_agent.clone(update={"model": ReplayLlm(model="replay", turns=list(recorded("adk", question)["model_turns"]))})
    return record_runs.printed(agent_adk.show, asyncio.run(agent_adk.ask(question, agent)))


def replay_langgraph(question):
    turns = recorded("langgraph", question)["model_turns"]
    messages = iter([AIMessage(content=t["content"], tool_calls=[{**c, "id": f"call-{i}-{j}"} for j, c in enumerate(t["tool_calls"])])
                     for i, t in enumerate(turns)])
    return record_runs.printed(agent_langgraph.show, asyncio.run(agent_langgraph.ask(question, ReplayChatModel(messages=messages))))


class AgentReplay(unittest.TestCase):
    def test_adk_agent_resolves_a_task_name_to_its_id(self):
        transcript = replay_adk("Close the docs task.")
        self.assertEqual(transcript, recorded("adk", "Close the docs task.")["transcript"])
        self.assertIn("call   complete_task {'task_id': 'T-3'}", transcript)

    def test_langgraph_agent_resolves_a_task_name_to_its_id(self):
        transcript = replay_langgraph("Close the docs task.")
        self.assertEqual(transcript, recorded("langgraph", "Close the docs task.")["transcript"])
        self.assertIn("call   complete_task {'task_id': 'T-3'}", transcript)

    def test_adk_returns_a_tool_error_to_the_model(self):
        transcript = replay_adk("Mark task T-9 as done.")
        self.assertEqual(transcript, recorded("adk", "Mark task T-9 as done.")["transcript"])
        self.assertIn("result error Error executing tool complete_task: No task with ID T-9.", transcript)

    def test_langgraph_returns_a_tool_error_to_the_model(self):
        transcript = replay_langgraph("Mark task T-9 as done.")
        self.assertEqual(transcript, recorded("langgraph", "Mark task T-9 as done.")["transcript"])
        self.assertIn("result error Error executing tool complete_task: No task with ID T-9.", transcript)

    def test_get_tools_starts_a_new_server_session_for_each_call(self):
        async def add_then_list():
            tools = {tool.name: tool for tool in await agent_langgraph.client.get_tools()}
            await tools["add_task"].ainvoke({"title": "Tag v1.2"})
            return await tools["list_tasks"].ainvoke({})

        self.assertNotIn("Tag v1.2", str(asyncio.run(add_then_list())))

    def test_one_session_keeps_server_state_between_calls(self):
        async def add_then_list():
            async with agent_langgraph.client.session("tasks") as session:
                tools = {tool.name: tool for tool in await load_mcp_tools(session)}
                await tools["add_task"].ainvoke({"title": "Tag v1.2"})
                return await tools["list_tasks"].ainvoke({})

        self.assertIn("Tag v1.2", str(asyncio.run(add_then_list())))

    def test_langgraph_without_error_handling_raises_instead(self):
        async def call_unknown_task():
            async with agent_langgraph.client.session("tasks") as session:
                tools = {tool.name: tool for tool in await load_mcp_tools(session, handle_tool_errors=False)}
                try:
                    await tools["complete_task"].ainvoke({"task_id": "T-9"})
                except ToolException as error:
                    return error

        self.assertIn("No task with ID T-9", str(asyncio.run(call_unknown_task())))

    def test_both_frameworks_discover_the_recorded_tools(self):
        for name, module in [("adk", agent_adk), ("langgraph", agent_langgraph)]:
            self.assertEqual(record_runs.printed(lambda: asyncio.run(module.discover())), RECORD["discovery"][name])

    def test_every_recorded_trial_reached_the_expected_outcome(self):
        for run in RECORD["runs"]:
            for trial in run["trials"]:
                calls = trial["calls"]
                if run["question"] == "Close the docs task.":
                    done = calls.index("complete_task {'task_id': 'T-3'}")
                    self.assertTrue(calls[done + 1].startswith(("ok", "success")), calls)
                else:
                    self.assertEqual(calls[0], "complete_task {'task_id': 'T-9'}")
                    self.assertTrue(calls[1].startswith("error Error executing tool complete_task: No task with ID T-9"))

    def test_agent_server_is_the_typed_tools_lesson_server(self):
        first = (HERE.parent / "mcp-typed-tools" / "tasks_server.py").read_bytes()
        self.assertEqual((HERE / "tasks_server.py").read_bytes(), first)


if __name__ == "__main__":
    unittest.main()

Return MCP tool responses to the model

Both frameworks run the same loop. The model gets the conversation and the tool declarations, and answers with either text or a tool call. The framework sends each tool call as an MCP tools/call, turns the result into a message for the model, and asks the model again. The frameworks differ in that last conversion. ADK passes the MCP result through as the function response, isError field included:

{
  "content": [
    {
      "type": "text",
      "text": "Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs."
    }
  ],
  "isError": true
}

LangGraph turns the same result into a ToolMessage with status error and the error text as its content. In both cases the model reads the message the server wrote, which is why the ToolError text in the first lesson names the tool that returns valid IDs.

Check your understanding

Secure a remote MCP server with OAuth 2.1 and scoped tokens continues with the next part of the MCP application.

References

Official sources checked October 21, 2026, against the package versions in requirements-lock.txt.

• ADK: MCP tools
• ADK: Ollama models through LiteLLM
• langchain-mcp-adapters repository
• LangChain: MCP and LangChain: agents (create_agent)
• LangChain: ChatOllama
• MCP specification: tools, error handling
• Gemma 4 model card and gemma4 on Ollama