Google ADK - Mesh API

Google’s Agent Development Kit (ADK) is a code-first framework for building agents: you define an LlmAgent, give it Python functions as tools, and run it locally with adk web or deploy it as a container. ADK talks to Gemini natively and reaches every other model through LiteLLM. Mesh plugs into that second path. Because Mesh speaks the OpenAI chat-completions dialect, a single LiteLlm connector points your ADK agents at Mesh — and from there at every model in the catalog, with your existing spend caps, rate limits, and usage logging. No Mesh-specific code goes into your agent.

Set the model to auto and Mesh’s auto-router picks the model per request. Your ADK agent gets routing, fallback, and cost control without choosing a model at all.

How it connects

LlmAgent → LiteLlm("openai/<model>") → https://api.meshapi.ai/v1 → Mesh routes → provider

The openai/ prefix selects LiteLLM’s OpenAI-dialect transport. LiteLLM strips the prefix before sending, so Mesh receives the bare model name — auto, or any catalog id — and routes it like any other API call.

Install

ADK needs Python 3.10 or newer. Install ADK, LiteLLM, and the mcp package together:

pip install "google-adk>=2.13.0" "litellm>=1.84" mcp

Two install traps, both of which produce confusing failures.mcp is a hard runtime dependency of LiteLLM’s dispatch path, but it is not always pulled in. Without it, every model call fails as APIConnectionError: OpenAIException - No module named 'mcp' — a connection error that has nothing to do with your network or your key.ADK requires litellm>=1.84. Versions 1.82.7 and 1.82.8 were compromised on PyPI (see the ADK security advisory). Never install those two. If you installed either, rotate your credentials.

A known-good pinned set:

requirements.txt

google-adk==2.13.0
litellm==1.106.4
mcp==2.3.0

Quickstart

ADK discovers agents by package layout. Create a directory whose __init__.py imports the agent module, and export a root_agent:

my_project/
├── mesh_agent/
│   ├── __init__.py      # from . import agent
│   └── agent.py         # defines root_agent
└── .env

mesh_agent/__init__.py

from . import agent

mesh_agent/agent.py

import os

from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm


def mesh_model(model: str = "auto") -> LiteLlm:
    """Build a Mesh-backed model connector for an ADK agent."""
    return LiteLlm(
        model=f"openai/{model}",                 # openai/ = OpenAI-dialect transport
        api_base="https://api.meshapi.ai/v1",
        api_key=os.environ["MESH_API_KEY"],      # your rsk_ key
    )


def get_shipment_status(order_id: str) -> dict:
    """Look up the shipment status of an order.

    Args:
        order_id: The order identifier, e.g. "A-1042".

    Returns:
        dict with a 'status' key ('success' or 'error'). On success also
        carries 'state' and 'eta_days'; on error an 'error_message'.
    """
    book = {
        "A-1042": {"state": "in transit", "eta_days": 2},
        "A-2190": {"state": "delivered", "eta_days": 0},
    }
    if order_id in book:
        return {"status": "success", "order_id": order_id, **book[order_id]}
    return {"status": "error", "error_message": f"No order {order_id}."}


root_agent = LlmAgent(
    name="mesh_support_agent",
    model=mesh_model(),
    description="Answers order-shipment questions, powered by Mesh.",
    instruction=(
        "You are a concise support agent. Use the get_shipment_status tool "
        "whenever the user asks about an order. State the state and the ETA."
    ),
    tools=[get_shipment_status],
)

.env

MESH_API_KEY=rsk_YOUR_KEY

Then run the ADK dev UI from the project root and talk to the agent in a browser:

export MESH_API_KEY=rsk_YOUR_KEY
adk web

Get your rsk_ key from Dashboard → API Keys. See Authentication for how rsk_ keys differ from mak_ admin keys.

Choosing the model

The string after openai/ is whatever Mesh should route. Both forms are valid:

Auto-router (recommended)
Pinned model

LiteLlm(
    model="openai/auto",
    api_base="https://api.meshapi.ai/v1",
    api_key=os.environ["MESH_API_KEY"],
)

Mesh picks the model per request, applies fallback, and optimises for your configured routing strategy. Your agent code never names a model. See Auto-routing and Routing algorithms.

LiteLlm(
    model="openai/openai/gpt-4o-mini",
    api_base="https://api.meshapi.ai/v1",
    api_key=os.environ["MESH_API_KEY"],
)

The first openai/ is LiteLLM’s transport selector; everything after it is the Mesh catalog id — so a model whose id already contains a slash reads as doubled. Browse ids with GET /v1/models or see Models.

Tool calling

ADK generates the JSON schema for each tool from your function’s signature and docstring, and emits it in the standard tools format that Mesh already accepts — so tool calling works with no extra configuration. Write a real docstring with an Args: section: it becomes the tool description the model reasons over. Keep tool functions returning a dict with an explicit status field. ADK passes the return value straight back to the model, and a model handles {"status": "error", "error_message": "..."} far better than a raised exception.

Streaming

Pass a RunConfig with SSE streaming to get incremental output:

import asyncio

from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from mesh_agent.agent import root_agent

APP, USER, SESSION = "my_app", "user_1", "s1"


async def main() -> None:
    session_service = InMemorySessionService()
    await session_service.create_session(
        app_name=APP, user_id=USER, session_id=SESSION
    )
    runner = Runner(agent=root_agent, app_name=APP, session_service=session_service)

    message = types.Content(
        role="user",
        parts=[types.Part(text="What's the shipment status of order A-1042?")],
    )

    async for event in runner.run_async(
        user_id=USER,
        session_id=SESSION,
        new_message=message,
        run_config=RunConfig(streaming_mode=StreamingMode.SSE),
    ):
        # Skip partials when handling tool calls — see the warning below.
        if not event.partial:
            for call in event.get_function_calls() or []:
                print("tool call:", call.name, dict(call.args or {}))
        if event.is_final_response() and event.content:
            print("".join(p.text or "" for p in event.content.parts))


asyncio.run(main())

In SSE mode ADK surfaces each partial tool-call delta as its own event, so event.get_function_calls() fires repeatedly with incomplete or empty arguments before the complete call arrives. ADK itself executes the tool only once. But if you handle function-call events yourself, filter on event.partial as above — otherwise you will fire your tool many times with empty arguments.

Deploy

ADK ships a FastAPI app factory, so the agent runs anywhere that runs a container — Cloud Run, GKE, Agent Runtime, or plain Docker.

main.py

import os

from google.adk.cli.fast_api import get_fast_api_app

app = get_fast_api_app(
    agents_dir=os.path.dirname(os.path.abspath(__file__)),
    session_service_uri=os.environ.get("SESSION_SERVICE_URI"),
    allow_origins=[o for o in os.environ.get("ALLOW_ORIGINS", "").split(",") if o],
    web=os.environ.get("SERVE_WEB_UI", "true").lower() == "true",
)

Dockerfile

FROM python:3.12-slim

ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
WORKDIR /app

COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py ./
COPY mesh_agent ./mesh_agent

ENV PORT=8080
EXPOSE 8080

RUN useradd -m appuser && chown -R appuser /app
USER appuser

CMD exec uvicorn main:app --host 0.0.0.0 --port ${PORT}

Two things to settle before this serves real traffic.Pass MESH_API_KEY in as a secret at deploy time — never bake it into the image or commit it. On Cloud Run, store it in Secret Manager and mount it as an environment variable.get_fast_api_app defaults to an in-memory session service, which loses every conversation whenever the service restarts or a new revision rolls out. Set SESSION_SERVICE_URI to a real backend (a database URI, or an Agent Engine session service) before users depend on conversation history.

Known limitation: thought signatures

Mesh returns a thought_signature on tool calls from Gemini thinking models, which the provider requires echoed back verbatim on the following turn to keep its reasoning chain intact. LiteLLM’s OpenAI-dialect transport does not carry that field, so on a multi-turn tool call it is dropped and the model loses its thinking after the first turn. The agent still works and tools still run — the reasoning quality degrades quietly rather than failing. Until this is resolved, pin ADK agents to a non-thinking model rather than relying on auto if your workload depends on multi-turn tool reasoning:

LiteLlm(
    model="openai/openai/gpt-4o-mini",
    api_base="https://api.meshapi.ai/v1",
    api_key=os.environ["MESH_API_KEY"],
)

Agents that use tools in a single turn, or no tools at all, are unaffected.

Troubleshooting

Symptom	Cause	Fix
`APIConnectionError: OpenAIException - No module named 'mcp'`	LiteLLM’s dispatch path needs `mcp`, which is missing	`pip install mcp`
`NotFoundError` with an HTML body	`api_base` is wrong, or points at a host that is not Mesh	Use exactly `https://api.meshapi.ai/v1`
`401` / `AuthenticationError`	Missing or invalid key	Check `MESH_API_KEY` holds an `rsk_` key; see Authentication
Model id not found	The id after `openai/` is not a Mesh catalog id	List ids with `GET /v1/models`, or use `auto`
Your tool fires repeatedly with empty arguments	Partial SSE tool-call deltas are being handled	Filter on `event.partial` — see Streaming
`UnicodeDecodeError` on Windows	LiteLLM reads cached files as `cp1252`	Set `PYTHONUTF8=1`

Auto-routing — how Mesh picks a model for auto
Tool calling — the tool format ADK emits
MCP Server — connect an MCP-compatible agent instead
Authentication — rsk_ keys versus mak_ admin keys

​How it connects

​Install

​Quickstart

​Choosing the model

​Tool calling

​Streaming

​Deploy

​Known limitation: thought signatures

​Troubleshooting

​Related

How it connects

Install

Quickstart

Choosing the model

Tool calling

Streaming

Deploy

Known limitation: thought signatures

Troubleshooting

Related