adk integrating with mcp server
ADK provides integration with MCP server. This is an example of how agent can use mcp tools
import logging
import os
from dotenv import load_dotenv
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
load_dotenv()
SYSTEM_INSTRUCTION = (
"You are a specialized assistant for currency conversions. "
"Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
"If the user asks about anything other than currency conversion or exchange rates, "
"politely state that you cannot help with that topic and can only assist with currency-related queries. "
"Do not attempt to answer unrelated questions or use tools for other purposes."
)
logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
logger.info("--- 🤖 Creating ADK Currency Agent... ---")
root_agent = LlmAgent(
model="gemini-2.5-flash",
name="currency_agent",
description="An agent that can help with currency conversions",
instruction=SYSTEM_INSTRUCTION,
tools=[
MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=os.getenv("MCP_SERVER_URL", "http://localhost:8080/mcp")
)
)
],
)
# Make the agent A2A-compatible
a2a_app = to_a2a(root_agent, port=10000)
from google.adk.apps import App
app = App(root_agent=root_agent, name="currency_agent")
And the MCP server uses fastMCP and code is pretty straight forward, just like how you would do it via mcp tool
import asyncio
import logging
import os
import httpx
from fastmcp import FastMCP
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
mcp = FastMCP("Currency MCP Server 💵")
@mcp.tool()
def get_exchange_rate(
currency_from: str = "USD",
currency_to: str = "EUR",
currency_date: str = "latest",
):
"""Use this to get current exchange rate.
Args:
currency_from: The currency to convert from (e.g., "USD").
currency_to: The currency to convert to (e.g., "EUR").
currency_date: The date for the exchange rate or "latest". Defaults to "latest".
Returns:
A dictionary containing the exchange rate data, or an error message if the request fails.
"""
logger.info(
f"--- 🛠️ Tool: get_exchange_rate called for converting {currency_from} to {currency_to} ---"
)
try:
response = httpx.get(
f"https://api.frankfurter.app/{currency_date}",
params={"from": currency_from, "to": currency_to},
)
response.raise_for_status()
data = response.json()
if "rates" not in data:
logger.error(f"❌ rates not found in response: {data}")
return {"error": "Invalid API response format."}
logger.info(f"✅ API response: {data}")
return data
except httpx.HTTPError as e:
logger.error(f"❌ API request failed: {e}")
return {"error": f"API request failed: {e}"}
except ValueError:
logger.error("❌ Invalid JSON response from API")
return {"error": "Invalid JSON response from API."}
if __name__ == "__main__":
logger.info(f"🚀 MCP server started on port {os.getenv('PORT', '8080')}")
# Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
asyncio.run(
mcp.run_async(
transport="http",
host="0.0.0.0",
port=os.getenv("PORT", "8080"),
)
)
Comments