自定义 MCP Server

import asyncio
import os
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams

async def main():
    toolset = McpToolset(
        connection_params=StreamableHTTPConnectionParams(
            url=os.environ["MCP_SERVER_URL"],
            headers={"Authorization": f"Bearer {os.environ['MCP_SERVER_TOKEN']}"},
        ),
    )
    try:
        print([tool.name for tool in await toolset.get_tools()])
    finally:
        await toolset.close()

asyncio.run(main())

from mcp.server.fastmcp import FastMCP

server = FastMCP("calculator")

@server.tool()
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

if __name__ == "__main__":
    server.run(transport="stdio")

import asyncio
import sys
from pathlib import Path
from mcp import StdioServerParameters
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams

async def main():
    toolset = McpToolset(connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command=sys.executable,
            args=[str(Path(__file__).with_name("local_mcp_server.py"))],
        ),
        timeout=10.0,
    ))
    try:
        print([tool.name for tool in await toolset.get_tools()])
    finally:
        await toolset.close()

asyncio.run(main())