Skip to main content
The Google Agent Development Kit (ADK) supports MCP servers, enabling Python agents to use MCP tools and expose ADK tools via MCP.

​
Installation

pip install google-genai

​
Basic MCP Integration

Use MCP servers in ADK agents: 
import asyncio
from google import genai
from google.genai import types

async def main():
    client = genai.Client(
        vertexai=False,
        api_key="your-gemini-api-key"
    )

    # Connect to MCP server
    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/github-a1b2c3/mcp",
                "headers": {"Authorization": "Bearer your-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        # Use MCP tools through the agent
        await session.send("Get info about the vercel/ai repository", end_of_turn=True)

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(main())

​
MCP Toolbox for Databases

The MCP Toolbox provides pre-built MCP servers for 40+ databases:

​
BigQuery Example

import asyncio
from google import genai
from google.genai import types

async def main():
    client = genai.Client(api_key="your-gemini-api-key")

    # Connect to MCP Toolbox BigQuery server
    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/toolbox-bigquery/mcp",
                "headers": {"Authorization": "Bearer your-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        # Query your BigQuery data
        await session.send(
            "What are the top 10 products by revenue in the sales dataset?",
            end_of_turn=True
        )

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(main())

​
PostgreSQL Example

from google import genai
from google.genai import types

async def query_postgres():
    client = genai.Client(api_key="your-gemini-api-key")

    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/toolbox-postgres/mcp",
                "headers": {"Authorization": "Bearer your-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        await session.send(
            "Show me the schema for the users table",
            end_of_turn=True
        )

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(query_postgres())

​
Supported Databases

MCP Toolbox includes connectors for: Google Cloud:
  • BigQuery, AlloyDB, Spanner, Cloud SQL (PostgreSQL, MySQL, SQL Server)
  • Firestore, Bigtable, Dataplex, Cloud Monitoring
Relational:
  • PostgreSQL, MySQL, SQL Server, SQLite, ClickHouse, TiDB, YugabyteDB
NoSQL:
  • MongoDB, Couchbase, Redis, Valkey, Cassandra
Graph:
  • Neo4j, Dgraph
Analytics:
  • Looker, Trino

​
FastMCP Server Integration

Use FastMCP servers with ADK: 
import asyncio
from google import genai
from google.genai import types

async def main():
    client = genai.Client(api_key="your-gemini-api-key")

    # Connect to FastMCP server
    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/custom-fastmcp/mcp",
                "headers": {"Authorization": "Bearer your-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        await session.send("Use the custom tools", end_of_turn=True)

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(main())

​
Google Cloud Genmedia MCP Servers

Access Google’s generative media services: 
from google import genai
from google.genai import types

async def use_genmedia():
    client = genai.Client(api_key="your-gemini-api-key")

    # Connect to Genmedia MCP servers (Imagen, Veo, Chirp, Lyria)
    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/genmedia-imagen/mcp",
                "headers": {"Authorization": "Bearer your-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        await session.send(
            "Generate an image of a sunset over mountains",
            end_of_turn=True
        )

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(use_genmedia())

​
Multiple MCP Servers

Combine multiple MCP servers: 
import asyncio
from google import genai
from google.genai import types

async def main():
    client = genai.Client(api_key="your-gemini-api-key")

    mcp_config = types.LiveConnectConfig(
        mcp_servers=[
            {
                "url": "https://mcp.runlayer.com/github-a1b2c3/mcp",
                "headers": {"Authorization": "Bearer github-token"}
            },
            {
                "url": "https://mcp.runlayer.com/toolbox-bigquery/mcp",
                "headers": {"Authorization": "Bearer bq-token"}
            }
        ]
    )

    async with client.aio.live.connect(
        model="gemini-2.0-flash-exp",
        config=mcp_config
    ) as session:
        await session.send(
            "Get the latest issues from our repo and cross-reference with usage data from BigQuery",
            end_of_turn=True
        )

        async for response in session.receive():
            if response.text:
                print(response.text)

asyncio.run(main())

​
Resources