65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""
|
|
mcp_tools.py — MCP toolset integrations for OSVauco ADK agent.
|
|
Supports:
|
|
- Google BigQuery MCP (via StreamableHTTP, uses ADC/OAuth)
|
|
- Google Maps MCP (via StreamableHTTP, uses API key)
|
|
- Local stdio MCP server (dev/test only)
|
|
|
|
Requires: google-adk >= 2.0.0, google-auth
|
|
"""
|
|
|
|
import os
|
|
import google.auth
|
|
import google.auth.transport.requests
|
|
from google.adk.tools.mcp_tool.mcp_toolset import (
|
|
MCPToolset,
|
|
StdioServerParameters,
|
|
StreamableHTTPConnectionParams,
|
|
)
|
|
|
|
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
|
|
|
|
# --- BigQuery MCP (Google managed, OAuth) ---
|
|
BIGQUERY_MCP_URL = f"https://bigquery.googleapis.com/mcp/projects/{PROJECT_ID}"
|
|
|
|
def get_bigquery_mcp_toolset() -> MCPToolset:
|
|
"""Connect to Google's managed BigQuery MCP server using ADC (no key file)."""
|
|
credentials, _ = google.auth.default(
|
|
scopes=["https://www.googleapis.com/auth/bigquery"]
|
|
)
|
|
credentials.refresh(google.auth.transport.requests.Request())
|
|
return MCPToolset(
|
|
connection_params=StreamableHTTPConnectionParams(
|
|
url=BIGQUERY_MCP_URL,
|
|
headers={
|
|
"Authorization": f"Bearer {credentials.token}",
|
|
"x-goog-user-project": PROJECT_ID,
|
|
},
|
|
)
|
|
)
|
|
|
|
# --- Google Maps MCP (API key from env/Secret Manager) ---
|
|
MAPS_MCP_URL = "https://maps.googleapis.com/mcp"
|
|
|
|
def get_maps_mcp_toolset() -> MCPToolset:
|
|
"""Connect to Google's managed Maps MCP server using an API key."""
|
|
maps_api_key = os.environ.get("MAPS_API_KEY", "")
|
|
if not maps_api_key:
|
|
raise ValueError("MAPS_API_KEY env var not set. Add to Secret Manager and load at startup.")
|
|
return MCPToolset(
|
|
connection_params=StreamableHTTPConnectionParams(
|
|
url=MAPS_MCP_URL,
|
|
headers={"X-Goog-Api-Key": maps_api_key},
|
|
)
|
|
)
|
|
|
|
# --- Local stdio MCP server (dev/test only) ---
|
|
def get_local_stdio_toolset(command: str, args: list[str]) -> MCPToolset:
|
|
"""Connect to a local stdio MCP server for development and testing."""
|
|
return MCPToolset(
|
|
connection_params=StdioServerParameters(
|
|
command=command,
|
|
args=args,
|
|
)
|
|
)
|