Connect an agent to MCP servers with McpToolset

When you need this. Another team already maintains a task tracker as an MCP server. Your docuctl assistant needs to find and add documentation tasks, but should not duplicate the server’s tools or expose its task-completion operation. We will connect that server through ADK’s McpToolset and select the two tools this assistant needs.

MCP is the protocol between the agent application and the tool server. McpToolset is the client adapter: it discovers the server’s tools and makes their declarations available to the model. The server still owns their implementation.

Set up the environment

Download the lab using Download code and open adk-mcp-toolset-lesson. It contains one Python client, tasks_server.py, documentation and offline tests. The tracker holds sample tasks in memory; it is not connected to an external issue service.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
cp .env.example .env

Set your Gemini API key in the private .env. agent.py loads it before constructing the agent. The MCP extra supplies the server SDK; both parent and child use the activated Python environment.

First inspect the documentation that the assistant can read:

# Roll back a release

`docuctl rollback` moves an environment back to an earlier release. It does not rebuild anything; it points traffic at the release you name.

    docuctl rollback --env production --to r41

Without `--to`, docuctl rolls back to the previous release. A rollback is itself recorded as a new release, so `docuctl releases` shows it.

Database migrations are not rolled back. If the release you leave ran a migration, write a new migration that reverses it.

docuctl 1.x used `docuctl release revert <release>`. That command was removed in docuctl 2.0.

The server keeps three sample tasks in TASKS. Its list_tasks, add_task and complete_task functions have typed arguments and results:

class Task(BaseModel):
    """One task in the tracker."""
    id: str = Field(description="Task ID such as T-1.")
    title: str
    priority: Priority
    status: Status


class TaskList(BaseModel):
    """The tasks that match a filter."""
    tasks: list[Task]
    count: int


server = MCPServer("tasks", version="1.0.0", instructions="Read and update the task tracker of one software project.")

TASKS: dict[str, Task] = {task.id: task for task in [
    Task(id="T-1", title="Write the release notes", priority="normal", status="open"),
    Task(id="T-2", title="Fix the login redirect bug", priority="high", status="done"),
    Task(id="T-3", title="Update the API docs", priority="low", status="open"),
]}

Before making a model request, check discovery:

python agent.py --tools

The filtered discovery in the supplied example lists:

Recorded example. The archived “Connect an agent to MCP servers with McpToolset” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.

add_task
list_tasks

This command starts and closes the real local MCP child. Do not start a second server terminal: ADK owns the child’s stdin and stdout. Server diagnostics belong on stderr so they do not corrupt the protocol.

Attach the server with McpToolset

The client factory defines the process command, timeout and allowed tool names:

def tracker(tool_filter: list[str] | None = ("list_tasks", "add_task"), python: str | None = None) -> McpToolset:
    """The task tracker MCP server as a child process; only the tools in tool_filter reach the model."""
    command = python or os.environ.get("TASKS_SERVER_PYTHON", sys.executable)
    return McpToolset(
        connection_params=StdioConnectionParams(
            server_params=StdioServerParameters(command=command, args=[str(HERE / "tasks_server.py")]),
            timeout=10,  # setup and default MCP request-response timeout in this SDK
        ),
        tool_filter=list(tool_filter) if tool_filter else None,
    )


def build_agent(model=None, toolset: McpToolset | None = None) -> Agent:
    return Agent(
        name="docs_assistant",
        model=model or (LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL),
        instruction="""You help developers keep the docuctl documentation correct.
Read pages with read_page. Track docs work in the task tracker: list the tasks before you add one,
and do not add a task that is already open. Say plainly when you cannot do something with your tools.""",
        tools=[read_page, toolset or tracker()],
        generate_content_config=types.GenerateContentConfig(
            temperature=0.2, max_output_tokens=1024, thinking_config=types.ThinkingConfig(thinking_level="low")),
    )

StdioServerParameters launches tasks_server.py with this environment’s Python. StdioConnectionParams adds a ten-second setup and request-response timeout. The process starts when the toolset first needs to discover tools. TASKS_SERVER_PYTHON is an optional override if your child interpreter must differ.

The agent combines its local read_page function with the remote tracker tools. Model responses request LOW thinking and at most 1,024 output tokens.

A server that cannot start. Tool discovery inside the agent can log an error and continue without the MCP tools. Check get_tools() explicitly at application startup when the toolset is required, and stop startup if discovery raises. The --tools command provides that check for this lab.

A tool excluded by tool_filter. complete_task is absent from this agent’s declarations. This is a client filter, not server authorization: another client with server access could still call that operation. Enforce shared permissions on the server.

A hostile tool description. ADK fences server-provided descriptions as untrusted data before presenting them to the model. That helps separate tool metadata from instructions, but does not make an arbitrary server trustworthy. Select the tools and servers deliberately.

Run the agent

Ask for a task with an explicit title and priority:

python agent.py "List the open tasks. If none already covers it, add a normal-priority task titled Clarify rollback migration warning."

Look for list_tasks, followed by add_task when no matching task exists. The output distinguishes model tool calls from server results. The new task belongs to this child process; a new command starts a fresh tracker with the original sample tasks.

Try the excluded operation in a separate run:

python agent.py "Mark task T-1 as done."

The assistant should explain that it has no completion tool. Compare that explanation with the discovered tool list rather than assuming the server rejected the call. This lesson exposes a local write without the previous lesson’s confirmation gate; add such a gate before adapting it to a tracker where writes need approval.

Adapt MCP tools into ADK calls

The model proposes a function call. ADK’s MCP adapter sends a tools/call request to the child, then returns the server’s CallToolResult as the function response. The result can contain text, typed structuredContent and an isError flag. The next model turn uses that result to explain what happened.

The Runner’s lifecycle also owns the connection:

async def ask(text: str, agent: Agent | None = None) -> list:
    """Run one question in a new session; closing the runner also stops the MCP server process."""
    sessions = InMemorySessionService()
    runner = Runner(app_name="docs", agent=agent or build_agent(), session_service=sessions)
    try:
        session = await sessions.create_session(app_name="docs", user_id="dev")
        message = types.Content(role="user", parts=[types.Part(text=text)])
        return [e async for e in runner.run_async(user_id="dev", session_id=session.id, new_message=message)]
    finally:
        await runner.close()

Closing the runner in finally releases the toolset and its child process on both success and failure. For a shared server, ADK also supports Streamable HTTP; the server then owns its own process lifecycle.

Test the agent against the real server

python -m unittest -v test_agent.py output_cap_check.py

The tests substitute saved model turns while running the real MCP client, stdio child and server. They check discovery, filtered declarations, result forwarding, failed startup and description fencing without a provider request.

Next, persist session state and separate it from memory so a returning developer can retain preferences between conversations.

References

• ADK MCP tools: toolsets, transports and filtering.
• MCP tools specification: discovery, calls and results.
• ADK runtime: Runner and resource lifecycle.