How do I call Ballet playbooks from Google ADK?
TL;DR: Use the Agent Development Kit's McpToolset with StreamableHTTPConnectionParams to connect to Ballet's MCP endpoint, or define a FunctionTool that calls the execute endpoint. Point the toolset at https://app.ballet.dev/mcp with an Authorization header and add it to your agent's tools.
Who this is for
Python developers building agents with Google's Agent Development Kit (ADK).
Option A: Connect with McpToolset
Attach an McpToolset configured for remote Streamable HTTP. The toolset imports Ballet's tools and manages the connection lifecycle.
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
import os
ballet_tools = McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://app.ballet.dev/mcp",
headers={"Authorization": f"Bearer {os.environ['BALLET_API_TOKEN']}"},
),
# optional: restrict which tools the agent can call
tool_filter=["list_playbooks", "run_playbook", "get_run"],
)
root_agent = LlmAgent(
model="gemini-2.5-flash",
name="ops_agent",
instruction="You run Ballet playbooks to complete operational tasks.",
tools=[ballet_tools],
)
Use McpToolset (lowercase "c"); the older MCPToolset spelling is deprecated. For per-request auth, pass a header_provider callable instead of static headers.
Option B: Define one playbook as a FunctionTool
To expose a single playbook, wrap the REST execute endpoint in a function and register it as a tool.
import json, os, httpx
from google.adk.agents import LlmAgent
async def run_onboarding(customer_id: str) -> dict:
"""Run the customer onboarding playbook in Ballet."""
url = f"https://app.ballet.dev/api/playbooks/{os.environ['BALLET_PLAYBOOK_ID']}/execute"
headers = {
"Authorization": f"Bearer {os.environ['BALLET_API_TOKEN']}",
"Content-Type": "application/json",
}
run = None
async with httpx.AsyncClient(timeout=None) as http:
async with http.stream("POST", url, headers=headers,
json={"input": {"customerId": customer_id}}) as res:
async for line in res.aiter_lines():
if line.startswith("data:"):
event = json.loads(line[5:].strip())
if event.get("type") == "run_stop":
run = event["run"]
return {"success": run and run.get("success"), "output": run and run.get("output")}
root_agent = LlmAgent(
model="gemini-2.5-flash",
name="onboarding_agent",
instruction="Use run_onboarding to onboard new customers.",
tools=[run_onboarding],
)
ADK wraps a plain function as a tool automatically. See Run playbooks over the REST API for event details.
Which option should I use?
- McpToolset — expose all (or a filtered set of) playbooks to the agent.
- FunctionTool — expose one playbook with an explicit signature.