A small guided ADK and MCP example

←Back to the MCP and ADK article

This tutorial mirrors the shape of the Crypto Market Explorer: an MCP service retrieves a narrow piece of public evidence, while a Google ADK agent turns that evidence into a short, factual response.

An MCP server makes a capability available to a compatible client. The useful boundary is not “everything the provider can do”; it is one understandable operation with validated input and a predictable result. In the excerpt, MarketDataService owns validation and the provider call; the MCP tool only exposes its safe, read-only result.

# market_mcp/mcp_server.py
from collections.abc import Awaitable, Callable
from typing import Any

from mcp.server import MCPServer
from .market_data import MarketDataError, MarketDataService

def create_mcp_server(market_data: MarketDataService) -> MCPServer:
    mcp = MCPServer(name="market-demo")

    @mcp.tool()
    async def get_market_snapshot(symbol: str) -> dict[str, Any]:
        """Return a read-only price snapshot for one supported symbol."""
        return await safe_tool_call(lambda: market_data.market_snapshot(symbol))

    return mcp

async def safe_tool_call(operation: Callable[[], Awaitable[dict[str, Any]]]) -> dict[str, Any]:
    try:
        return {"ok": True, "data": await operation()}
    except (ValueError, MarketDataError) as error:
        return {"ok": False, "error": {"message": str(error)}}

The server owns the provider call. The agent never receives an exchange API key, account permission, or unrestricted HTTP client. It receives a named, read-only tool. The runnable MCP project also mounts this server on a Streamable HTTP endpoint.

Before an agent is involved, the application decides what is allowed. A small allowlist makes the experience clearer for the visitor and easier to test for the team.

from dataclasses import dataclass

@dataclass(frozen=True)
class GuidedQuestion:
    title: str
    tool_name: str
    tool_arguments: dict[str, object]

SUPPORTED_MARKETS = {"BTC", "ETH", "SOL"}

def price_question(market: str) -> GuidedQuestion:
    if market not in SUPPORTED_MARKETS:
        raise ValueError("Choose a supported market.")
    return GuidedQuestion(
        title=f"{market} price",
        tool_name="get_market_snapshot",
        tool_arguments={"symbol": f"{market}USDT"},
    )

A real interface can present this as a selector rather than a text box. The important part is that the browser sends a known question and known market—not a free-form instruction.

The application’s MCP client retrieves the approved data before the model can see it. The agent gets one tool created for that guided question, can retrieve the evidence once, and then uses the returned values to write the answer.

# market_agent/adk_answerer.py
from collections.abc import Awaitable, Callable
from typing import Any
from uuid import uuid4

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

async def answer_guided_question(
    *, user_id: str, question: GuidedQuestion, retrieve: Callable[[], Awaitable[dict[str, Any]]]
) -> str:
    tool_used = False

    # Expose one zero-argument ADK tool for this question's evidence only.
    async def market_tool() -> dict:
        """Retrieve the evidence required for this guided market question."""
        nonlocal tool_used
        if tool_used:
            raise RuntimeError("This guided tool can be used once.")
        tool_used = True
        return await retrieve()

    # ADK uses the function name as the tool name the model can call.
    market_tool.__name__ = question.tool_name
    agent = LlmAgent(
        name="guided_market_agent",
        model="gemini-2.5-flash",
        instruction="Use the supplied market evidence. Be factual and concise.",
        tools=[market_tool],
    )
    # Create a short-lived in-memory session for this one guided run.
    sessions = InMemorySessionService()
    session_id = str(uuid4())
    await sessions.create_session(app_name="market_demo", user_id=user_id, session_id=session_id)
    runner = Runner(agent=agent, app_name="market_demo", session_service=sessions)
    message = types.Content(role="user", parts=[types.Part(text=question.title)])

    # Return the final response from ADK's streamed events.
    async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=message):
        if event.is_final_response() and event.content and event.content.parts:
            return "".join(part.text or "" for part in event.content.parts).strip()
    raise RuntimeError("The agent did not return a response.")

The one-use guard is intentional. It does not limit the MCP service’s internal work: a comparison tool may still retrieve several market snapshots. It prevents the model from repeatedly asking the same approved agent tool for another answer.

This example deliberately stops before authentication, provider credentials, rate limits, caching, history, deployment, and monitoring. Those are important in a real service, but they obscure the first concept: a guided agent is useful because its job and evidence are deliberately narrow.

Once this small path makes sense, add those safeguards around it—without widening the original question or tool boundary by accident.