---
title: "How do I call Ballet playbooks from Pydantic AI?"
description: "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:."
canonical_url: "https://docs.ballet.dev/articles/how-do-i-call-ballet-playbooks-from-pydantic-ai-K5Tei4NZRG"
md_url: "https://docs.ballet.dev/articles/how-do-i-call-ballet-playbooks-from-pydantic-ai-K5Tei4NZRG.md"
---
# 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.

```python
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.

```python
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](/articles/how-do-i-run-a-playbook-over-the-rest-api-KQIz0apagm) 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.

## Related articles

- [How do I use Ballet playbooks in an agent framework?](/articles/how-do-i-use-ballet-playbooks-in-an-agent-framework-q3PWTx18lY)
- [The MCP endpoint](/articles/how-do-i-connect-to-ballets-mcp-endpoint-1ydPKBzHZm)
- [Run playbooks over the REST API](/articles/how-do-i-run-a-playbook-over-the-rest-api-KQIz0apagm)