Skip to main content

Section 5: Build a Universal MCP Client

Companion Guide

This documentation serves as the written companion to the Master Google ADK video course. It aggregates content from Lectures 56 through 66.

Bestseller
Master Google ADK - Agent Development Kit

Master Google ADK - Agent Development Kit

Build Multi Agent Systems | Mac, Windows, Ubuntu | Deploy to Google Cloud | Free Gemini Key

★ 4.4👥 470+
View on Udemy

1. Why Build MCP Clients with Google ADK?​

The Model Context Protocol (MCP) landscape is evolving. The new Streamable HTTP transport allows MCP servers to be deployed remotely, replacing the older HTTP+SSE method. However, powerful AI agents often need to interact with local resources (like your filesystem) via STDIO processes.

We need a "Universal Client" capable of bridging this gap. By building this with the Google Agent Development Kit (ADK), we gain:

  • Unified Interface: A single LlmAgent that can talk to both stateless HTTP servers and stateful local processes simultaneously.
  • Type Safety: ADK provides robust typing for MCP toolsets.
  • Streaming: Native support for streaming thoughts and tool execution steps back to the user.

2. STEP 1 - Building MCP Servers for our ADK based MCP Client​

Before building the client, we need tools for it to consume. In this architecture, we utilize two distinct server types:

  1. Remote Math Servers (HTTP): Two separate servers running on localhost:3000 and localhost:3001. These are stateless servers built in previous sections that provide tools like add_numbers and multiply_numbers.
  2. Local Terminal Server (STDIO): A server we will build using FastMCP that runs directly on your machine to execute shell commands.

Ensure your HTTP servers are running before starting the client development:

# Example commands to start your existing HTTP servers
uv run streamable_http_server/main.py --server server1
uv run streamable_http_server/main.py --server server2

3. STEP 2 Building the ADK MCP Client​

The ADK MCP Client is not just a script; it is a layered application. We structure the project to separate concerns:

  • User Interface Layer: Handles input/output and streaming.
  • Client Layer: Manages session lifecycle and connection negotiation.
  • Agent Layer: Wraps the Gemini model and the specific toolsets.

This separation ensures that we can easily swap out the UI (e.g., changing from CLI to a Web UI) without rewriting the core agent logic.

4. Introduction, Preview of What We'll Build​

We are building a Chat Application that acts as an orchestration layer.

The Scenario: You will ask the agent: "Can you add 32 and 2, then write the output to a file named 'results.txt'?"

The Workflow:

  1. The Agent receives the prompt.
  2. It identifies it needs a math tool. It routes a request via HTTP to server1 to calculate 32 + 2 = 34.
  3. It receives the result.
  4. It identifies it needs to write a file. It routes a request via STDIO to the local terminal server to run an echo command.
  5. It confirms the action to you.

This demonstrates the "Universal" capability: combining remote logic with local side effects in a single turn.

5. Code overview​

Our project directory is structured as follows:

  • cmd.py: The entry point and Command Line Interface.
  • client.py: The MCPClient class definition.
  • agent.py: The AgentWrapper and LlmAgent initialization.
  • theialanguage_config.json: The configuration file defining server connections.
  • terminal_server.py: The local STDIO server implementation.
  • utilities.py: Helper functions for JSON parsing and pretty printing.

6. GOOGLE_API_KEY setup​

To use the Gemini 2.0 Flash model via Google ADK, you must configure your environment variables.

Create a .env file in your root directory:

.env
GOOGLE_API_KEY=your_actual_api_key_here

In our code, we use python-dotenv to load this automatically at runtime within utilities.py.

7. MCP Client User Interface - cmd.py​

The cmd.py file contains the main event loop. It is responsible for initializing the client and handling the asynchronous stream of events.

cmd.py
import asyncio
from client import MCPClient
# ... other imports

async def chat_loop():
    # 1. Initialize Client
    client = MCPClient(
        app_name="google-adk-gemini-mcp-client",
        user_id="theialanguage_001",
        session_id="session_001"
    )
    
    # 2. Negotiate Connections
    await client.init_session()

    print("🤖 ADK LLM Agent Chat Started.")

    while True:
        user_input = input("You: ")
        if user_input.lower() in ["quit", "q"]: break

        # 3. Stream Events
        async for event in await client.send_task(user_input):
            # Pretty print intermediate reasoning steps
            print_json_response(event, "Event")
            
            # Print final answer
            if hasattr(event, "is_final_response") and event.is_final_response():
                print(f"Agent: {event.content.parts[0].text}")

    await client.shutdown()

if __name__ == "__main__":
    asyncio.run(chat_loop())

8. MCP Client Implementation - client.py​

The MCPClient class in client.py orchestrates the session. It uses ADK's InMemorySessionService to track the conversation state.

Key responsibilities:

  1. Session Creation: Establishes a context for the interaction.
  2. Agent Building: Triggers the loading of tools defined in the config.
  3. Task Running: Wraps user input into ADK's Content objects and dispatches them to the Runner.
client.py
class MCPClient:
    def __init__(self, app_name, user_id, session_id):
        self.session_service = InMemorySessionService()
        self.agent_wrapper = AgentWrapper() # From agent.py
        self.runner = None
        # ... setup variables

    async def init_session(self):
        await self.session_service.create_session(...)
        await self.agent_wrapper.build() # Connect to tools
        
        # Initialize the ADK Runner
        self.runner = Runner(
            agent=self.agent_wrapper.agent,
            session_service=self.session_service,
            # ... params
        )

    async def send_task(self, user_input):
        new_message = Content(role="user", parts=[Part(text=user_input)])
        # Returns an async generator for streaming
        return self.runner.run_async(..., new_message=new_message)

9. MCP Agent Implementation - agent.py​

This is the core "brain" logic. The AgentWrapper dynamically loads toolsets based on the configuration file.

This class iterates through every server defined in config.json.

  • If type is HTTP: It creates StreamableHTTPServerParams.
  • If type is STDIO: It creates StdioServerParameters.

It then instantiates the LlmAgent:

agent.py
class AgentWrapper:
    async def build(self):
        # Load tools from config (HTTP + STDIO)
        self._toolsets = await self._load_toolsets()
        
        # Initialize Gemini 2.0 Flash Agent
        self.agent = LlmAgent(
            model="gemini-2.0-flash",
            name="enterprise_assistant",
            instruction="Assist the user with filesystem and MCP server tasks.",
            tools=self._toolsets # Inject universal tools
        )

10. Code - MCP config json file​

This JSON file is the heart of our "Universal" capability. It tells the client where to find tools.

theialanguage_config.json
{
  "mcpServers": {
    "server1": {
      "type": "http",
      "url": "http://localhost:3000/mcp/" 
    },
    "server2": {
      "type": "http",
      "url": "http://localhost:3001/mcp/"
    },
    "terminal": {
      "type": "stdio",
      "command": "/users/theialanguage/.local/bin/uv", 
      "args": [
        "run",
        "--directory",
        "/users/theialanguage/mcp/workspace",
        "terminal_server.py"
      ]
    }
  }
}
tip

Notice the terminal entry. It uses stdio and points to a local python script. This allows the client to spawn that script as a subprocess and communicate via standard input/output.

11. Code for Utilities & STDIO Server + Environment Setup​

The Local Server (terminal_server.py)​

We use FastMCP to quickly spin up the local capability required by the config above.

terminal_server.py
from mcp.server.fastmcp import FastMCP
import subprocess

mcp = FastMCP("terminal")

@mcp.tool()
async def run_command(command: str) -> str:
    """Run a terminal command inside the workspace."""
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

if __name__ == "__main__":
    mcp.run(transport="stdio")

Running the Project​

  1. Dependencies: Ensure you have google-adk, mcp, and standard async libraries installed.
  2. Sync: Run uv sync to install dependencies.
  3. Execute: 
    uv run cmd.py

You now have a fully functional Universal MCP Client!

What's Next?​

This concludes the currently available written documentation modules for the Google ADK Masterclass.

Next Section:

Coming Soon to Docs:

  • Section 7: Deploying Agents to Google Cloud
  • Section 8: No-Code Agents (YAML)
  • Section 9: ADK Visual Agent Builder

To continue your learning journey or view the video lectures for these upcoming sections, please visit the course roadmap.

Return to Course Roadmap →

Bestseller
Master Google ADK - Agent Development Kit

Master Google ADK - Agent Development Kit

Build Multi Agent Systems | Mac, Windows, Ubuntu | Deploy to Google Cloud | Free Gemini Key

★ 4.4👥 470+
View on Udemy