How do I call Ballet playbooks from Pydantic AI?
TL;DR: Register Ballet's MCP endpoint as a toolset with MCPServerStreamableHTTP, or define a plain function tool that calls the execute endpoint. Point the server at https://app.ballet.dev/mcp with an Authorization header, pass it to Agent(..., toolsets=[server]), and run inside async with agent:.
Who this is for
Python developers building agents with Pydantic AI.
Option A: Register Ballet as a toolset over MCP
Create an MCPServerStreamableHTTP pointed at Ballet's endpoint and pass it to the agent's toolsets. Each MCP server is a toolset, so Ballet's tools (run_playbook, get_run, list_playbooks, …) become available to the model automatically.
import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
server = MCPServerStreamableHTTP(
"https://app.ballet.dev/mcp",
headers={"Authorization": f"Bearer {os.environ['BALLET_API_TOKEN']}"},
)
agent = Agent("openai:gpt-4o", toolsets=[server])
async def main():
async with agent:
result = await agent.run(
"Run the lead-enrichment playbook for cus_123 and summarize the result."
)
print(result.output)
The async with agent: block opens and closes the MCP connection for you.
Option B: Define one playbook as a tool
To expose a single playbook, register a plain function tool that calls the REST execute endpoint and returns the run output.
import json, os, httpx
from pydantic_ai import Agent
agent = Agent("openai:gpt-4o")
@agent.tool_plain
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")}
See Run playbooks over the REST API for the event details.
Which option should I use?
- MCP — register every playbook in the workspace as a toolset.
- REST tool — expose one playbook with a typed function signature.