Java World Knowledge Sharing Portal: 2025

In the world of AI applications, building robust and scalable systems often means connecting different services and agents. This blog post will walk you through a practical example of how to integrate a FASTMCP server with Google ADK agents, running locally and communicating with a FASTMCP client that is also running on your machine. The goal is to show how you can create a modular system where agents can communicate with each other through a well-defined protocol.

The Architecture: A Two-Part System
Our setup consists of two main Python scripts, each with a distinct role:

restapi-mcp-adk-client.py (The FASTMCP Server): This script acts as our backend. It’s a FASTMCP server that exposes a single tool. This tool’s job is to fetch data from an external REST API, but it doesn’t do the work itself. Instead, it delegates that task to a Google ADK agent. This demonstrates how a FASTMCP tool can be a gateway to a more complex, agent-driven process.

restapi-mcp-adk-server.py (The FASTMCP Client): This is our frontend or the entry point for the user. It also contains an ADK agent, which is the one you will interact with. When you ask this agent to get data, it will intelligently decide to use a tool that connects to our FASTMCP server from the first script.

This architecture creates a chain of command: User -> Client ADK Agent -> FASTMCP Client -> FASTMCP Server -> Server ADK Agent -> External REST API.

Code Breakdown
Let’s look at the code for each component.

The FASTMCP Server with ADK Agent
This script sets up the server. The get_objects_by_id_using_adk_agent function is the core of our server’s toolset. It’s marked with @mcp.tool() and, importantly, it uses an ADK agent (call_mcp_server_agent) to handle the actual API call.
restapi-mcp-adk-client.py

from fastmcp import FastMCP, Context
import httpx
import asyncio
from fastmcp import Client
import google.genai as genai
from typing import Any
from google.genai import types
from dotenv import load_dotenv
 
load_dotenv()
 
from google.adk.agents import LlmAgent
from pydantic import BaseModel, Field
 
from google.adk.agents import Agent
from google.adk.events import Event
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
 
mcp = FastMCP(
    name="RESTful API Wrapper <img draggable="false" role="img" class="emoji" alt="" src="https://s0.wp.com/wp-content/mu-plugins/wpcom-smileys/twemoji/2/svg/1f310.svg">",
)
 
BASE_URL = "https://api.restful-api.dev/objects"
 
APP_NAME = "MCP_SERVER_WITH_ADK_AGENT"
USER_ID = "1234"
SESSION_ID = "session1234"
 
'''
async def get_objects_by_ids(ids: list[str], ctx: Context):
    query = "&".join([f"id={i}" for i in ids])
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}?{query}")
        print("get_objects_by_ids",resp.json())
    return resp.json()
         
'''
async def get_object_by_id(object_id: str):
   async with httpx.AsyncClient() as client:
        resp = await client.get(f"{BASE_URL}/{object_id}")
        print("get_object_by_id",resp.json())
        return resp.json()
 
call_mcp_server_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="assistant",
    description="This agent is used to send data to FASTMCP client",
    instruction="""Help user to fetch the data from the RESTAPI and send it to the FASTMCP Client.
    When the user asks to fetch data for a specific object ID, use the `get_object_by_id` tool and pass the ID to it.
    """,
    tools=[get_object_by_id],
)
 
# Session and Runner
async def setup_session_and_runner():
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    if call_mcp_server_agent:
        runner = Runner(agent=call_mcp_server_agent, app_name=APP_NAME, session_service=session_service)
    return session, runner
 
async def get_agent_async(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    session, 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:
        # You can uncomment the following line to see the full event flow for debugging
        # print(f"DEBUG Event: {event.model_dump_json(indent=2, exclude_none=True)}")
        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
            #print final_response
            print(f"final_response",final_response)
    return final_response
     
@mcp.tool()
async def get_objects_by_id_using_adk_agent(object_id: str,ctx: Context):
    print(f"object_id:::::::::::::::::",object_id)
    final_result = await get_agent_async(f"Fetch the data for object_id {object_id}, pass the id to get_object_by_id tool")
    return final_result
 
if __name__ == "__main__":
    #mcp.run(transport="streamable-http", host="127.0.0.1", port=8001, path="/mcp")
    mcp.run()

The FASTMCP Client with ADK Agent
This script is the entry point. Its ADK agent is configured with a tool called get_mcp_data. This tool is where the magic happens—it uses fastmcp.Client to make a call to the server we defined above.
restapi-mcp-adk-server.py

import asyncio
from fastmcp import Client
from typing import Any
from google.genai import types
from dotenv import load_dotenv
 
from google.adk.agents import LlmAgent
from pydantic import BaseModel, Field
 
from google.adk.agents import Agent
from google.adk.events import Event
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
 
APP_NAME = "CALL_MCP_SERVER"
USER_ID = "1234"
SESSION_ID = "session1234"
 
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("restapi-mcp-adk-server.py") as client:
        single = await client.call_tool("get_objects_by_id_using_adk_agent", {"object_id": object_id})
        print("Fetched single:", single)
        return single
         
call_mcp_server_adk_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],
)
     
# Session and Runner
async def setup_session_and_runner():
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=call_mcp_server_adk_agent, app_name=APP_NAME, session_service=session_service)
    return session, runner
  
async def get_agent_async(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    session, 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:
        # You can uncomment the following line to see the full event flow for debugging
        # print(f"DEBUG Event: {event.model_dump_json(indent=2, exclude_none=True)}")
        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}")

Putting It All Together: The Execution Flow
When you run restapi-mcp-adk-server.py, the following sequence of events occurs, as shown in your output:

The client script starts and sends the query “Fetch the data for object_id 2” to its ADK agent.

The client ADK agent recognizes that it needs to fetch data and calls its get_mcp_data tool.

The get_mcp_data tool, acting as a FASTMCP client, initiates a call to the FASTMCP server (restapi-mcp-adk-client.py). It specifically calls the tool named get_objects_by_id_using_adk_agent with the object_id of 2.

The server receives this call, and its ADK agent takes over. It executes its own internal tool, get_object_by_id, passing it the id of 2.

get_object_by_id makes an HTTP request to the external https://api.restful-api.dev/objects/2 endpoint.

The external API returns the data, which is then passed back up the chain.

Finally, the client ADK agent receives the response and presents the final, conversational output: “OK. I have fetched the data for object ID 2. The object’s name is Apple iPhone 12 Mini, 256GB, Blue and its data is None.”

This example showcases the power of combining these technologies to build a system where different agents and services can collaborate seamlessly through a standardized tool-calling protocol.

output:-
 
(.venv) C:\vscode-python-workspace\adkagent>python restapi-mcp-adk-client.py
C:\vscode-python-workspace\adkagent\.venv\Lib\site-packages\pydantic\_internal\_fields.py:198: UserWarning: Field name "config_type"
 in "SequentialAgent" shadows an attribute in parent "BaseAgent"
  warnings.warn(
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="OK. I have fetched the data for object ID 2. The object's nam
e is Apple iPhone 12 Mini, 256GB, Blue and its data is None.\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. The object's name is Apple iPhone 12 Mini, 2
56GB, Blue and its data is None.
 
 
--- Script Finished ---
Final returned value: OK. I have fetched the data for object ID 2. The object's name is Apple iPhone 12 Mini, 256GB, Blue and its da
ta is None.
 
 
(.venv) C:\vscode-python-workspace\adkagent>

source code :- https://github.com/shdhumale/app.git

Java World Knowledge Sharing Portal

Monday, August 11, 2025

Building a Modular System: Local FASTMCP Client & Server with ADK Agents

Total Pageviews

Blog Archive

Contributors