Java World Knowledge Sharing Portal: Bridging the Gap: Consuming an External FASTMCP Server with a Google ADK Agent

Building powerful, intelligent agents often requires them to interact with external services. This article demonstrates how to create a Google Agent Development Kit (ADK) agent that can communicate with an external FASTMCP (Fast Multi-Cloud Platform) server. We’ll use the agent as a “translator,” allowing it to call tools on a separate server to fetch data.

Before you start: Make sure you have a FASTMCP server running. If you need to set one up, please refer to this article and repository:
https://siddharathadhumale.blogspot.com/2025/05/creating-mcp-server-and-client-in.html

High-Level Overview

This project shows how a Google ADK agent can leverage an external FASTMCP server’s capabilities. Instead of executing the final action itself, our agent will delegate the task to the FASTMCP server.

The workflow looks like this:

A user gives a query to our ADK agent (e.g., “Fetch data for object 2”).
The ADK agent, an LlmAgent, understands the request and identifies that it needs to use its local tool: get_mcp_data.
The get_mcp_data tool acts as a bridge. It uses the fastmcp.Client to connect to the running FASTMCP server.
It then instructs the FASTMCP server to execute one of its own remote tools, get_object_by_id, and passes the required object ID.
The FASTMCP server runs its tool, fetches the data, and sends the result back to our client.
The ADK agent receives this result from the tool and presents it to the user as a final response.

Code Breakdown

Let’s dive into the Python code. We’ll create a file named connetSSEMCPServer.py.

1. The Tool: A Bridge to the MCP Server

The core of our client-server interaction is this asynchronous function. This is the tool that our ADK agent will call.

1

2

3

4

5

6

7

async def get_mcp_data(object_id: str) -> dict:
    """Fetches an object by its ID from the MCP server."""
    print(f"Tool 'get_mcp_data' called with object_id: {object_id}")
    async with Client("http://127.0.0.1:8000/mcp") as client:
        single = await client.call_tool("get_object_by_id", {"object_id": object_id})
        print("Fetched single:", single)
        return single

This function is defined as a tool that the local ADK agent can use.
Inside the function, we initialize fastmcp.Client, pointing it to our FASTMCP server’s URL.
The key line is await client.call_tool(...). This sends a request to the server, asking it to execute the tool named get_object_by_id and passing it the object_id.
The function waits for the server’s response and then returns the result.

2. The ADK Agent

Next, we define our LlmAgent. This agent is responsible for understanding the user’s intent and deciding which tool to use.

call_mcp_server_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="assistant",
    description="This agent is used to get data using FASTMCP client by calling the FASTMCP server ",
    instruction="""Help user to fetch the data from the FASTMCP Server using FASTMCP Client.
    When the user asks to fetch data for a specific object ID, use the `get_mcp_data` tool and pass the ID to it.
    """,
    tools=[get_mcp_data],
)

We use the powerful gemini-2.0-flash model.
The instruction prompt is critical. It tells the agent its purpose: to fetch data from the FASTMCP server. Crucially, it instructs the agent to use the get_mcp_data tool whenever a user asks to fetch data for a specific ID.
By adding our function to the tools list, we make it available for the agent to call when it decides it’s appropriate.

Putting it All Together

Here’s the full script, including the standard ADK setup to run our agent.

import warnings
import asyncio
from fastmcp import Client
from typing import Any
from google.genai import types
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from pydantic import BaseModel, Field
 
# Suppress a known Pydantic warning
warnings.filterwarnings(
    "ignore",
    message='Field name "config_type" in "SequentialAgent" shadows an attribute in parent "BaseAgent"',
    category=UserWarning,
    module="pydantic._internal._fields"
)
 
APP_NAME = "CALL_MCP_SERVER"
USER_ID = "1234"
SESSION_ID = "session1234"
 
# ... (Insert get_mcp_data and call_mcp_server_agent definitions here) ...
async def get_mcp_data(object_id: str) -> dict:
    """Fetches an object by its ID from the MCP server."""
    print(f"Tool 'get_mcp_data' called with object_id: {object_id}")
    async with Client("http://127.0.0.1:8000/mcp") as client:
        single = await client.call_tool("get_object_by_id", {"object_id": object_id})
        print("Fetched single:", single)
        return single
 
call_mcp_server_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="assistant",
    description="This agent is used to get data using FASTMCP client by calling the FASTMCP server ",
    instruction="""Help user to fetch the data from the FASTMCP Server using FASTMCP Client.
    When the user asks to fetch data for a specific object ID, use the `get_mcp_data` tool and pass the ID to it.
    """,
    tools=[get_mcp_data],
)
 
 
# Standard ADK setup and runner
async def setup_session_and_runner():
    session_service = InMemorySessionService()
    await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=call_mcp_server_agent, app_name=APP_NAME, session_service=session_service)
    return runner
 
# Agent invocation function
async def get_agent_async(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    runner = await setup_session_and_runner()
    events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
 
    final_response = "Agent did not produce a final response."
    async for event in events:
        if event.is_final_response() and event.content and event.content.parts:
            print(f"Potential final response from [{event.author}]: {event.content.parts[0].text}")
            final_response = event.content.parts[0].text
 
    return final_response
 
if __name__ == "__main__":
    final_result = asyncio.run(get_agent_async("Fetch the data for object_id 2"))
    print(f"\n--- Script Finished ---\nFinal returned value: {final_result}")

3. Execution and Output

When you run this script (after ensuring your FASTMCP server is active), you’ll see the agent’s full thought process in the console.

The ADK agent receives the query and decides to call the get_mcp_data tool.
The get_mcp_data function is executed, which in turn calls the FASTMCP server.
The FASTMCP server returns the requested data for object_id=2.
The ADK agent receives this result and formats it into a human-readable final response.

(.venv) C:\vscode-python-workspace\adkagent>python connetSSEMCPServer.py
Warning: there are non-text parts in the response: ['function_call'], returning concatenated text result from text parts. Check the full candidates.content.parts accessor to get the full model response.
Tool 'get_mcp_data' called with object_id: 2
Fetched single: CallToolResult(content=[TextContent(type='text', text='{\n  "id": "2",\n  "name": "Apple iPhone 12 Mini, 256GB, Blue",\n  "data": null\n}', annotations=None, meta=None)], structured_content=None, data=None, is_error=False)
Potential final response from [assistant]: OK. I have fetched the data for object_id 2. Here is the content:

{
“id”: “2”,
“name”: “Apple iPhone 12 Mini, 256GB, Blue”,
“data”: null
}

--- Script Finished ---
Final returned value: OK. I have fetched the data for object_id 2. Here is the content:

{
“id”: “2”,
“name”: “Apple iPhone 12 Mini, 256GB, Blue”,
“data”: null
}

This successful run demonstrates a powerful pattern for building agents: decentralizing their capabilities. By connecting your agent to an external server, you can scale its functionality without embedding every single tool directly in the agent’s code.