Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.runlayer.com/llms.txt

Use this file to discover all available pages before exploring further.

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