Python SDK

Fully-typed, async-ready Python client for the AgentOven control plane.

The Python SDK provides a Pythonic, fully-typed interface to the AgentOven control plane. Built on httpx and pydantic for async support and strong typing.

Requires Python 3.10+ · Install with pip install agentoven

Installation

terminal

$ pip install agentoven
# With async support
$ pip install agentoven[async]

Quick Start

python

from agentoven import AgentOven, Agent

# Connect to control plane
oven = AgentOven("http://localhost:8080")

# Register an agent
agent = Agent(
    name="research-bot",
    framework="langchain",
    skills=["web-search", "summarization"],
)
oven.agents.register(agent)

# Deploy
oven.agents.bake("research-bot")

# Send task via A2A
result = oven.a2a.send_task(
    agent="research-bot",
    message="Summarize AI safety papers",
)
print(result.text)

Async Usage

python

import asyncio
from agentoven import AsyncAgentOven, Agent

async def main():
    oven = AsyncAgentOven("http://localhost:8080")

    # Register and bake
    agent = Agent(name="fast-bot", framework="crewai", skills=["analysis"])
    await oven.agents.register(agent)
    await oven.agents.bake("fast-bot")

    # Stream response via SSE
    async for chunk in oven.a2a.send_task_stream(
        agent="fast-bot",
        message="Analyze quarterly sales data",
    ):
        print(chunk.text, end="", flush=True)

asyncio.run(main())

Managing Agents

python

from agentoven import AgentOven

oven = AgentOven("http://localhost:8080")

# List all agents
agents = oven.agents.list()
for a in agents:
    print(f"{a.name} — {a.status} — {a.tasks_completed} tasks")

# Get agent status
status = oven.agents.status("research-bot")
print(f"Uptime: {status.uptime}")
print(f"Health: {status.health}")

# Cool an agent
oven.agents.cool("research-bot")

# Re-bake
oven.agents.bake("research-bot")

Running Recipes

python

from agentoven import AgentOven

oven = AgentOven("http://localhost:8080")

# Run a recipe from file
result = oven.recipes.run("pipeline.yaml")
print(f"Recipe completed in {result.duration}s")
print(f"Steps: {len(result.steps)}")

# Run with parameters
result = oven.recipes.run(
    "pipeline.yaml",
    params={"topic": "AI safety", "depth": "detailed"},
)

Error Handling

python

from agentoven import AgentOven
from agentoven.exceptions import (
    AgentNotFoundError,
    TaskTimeoutError,
    RateLimitError,
)

oven = AgentOven("http://localhost:8080")

try:
    result = oven.a2a.send_task(
        agent="research-bot",
        message="Analyze data",
        timeout=30,
    )
except AgentNotFoundError:
    print("Agent not registered or not baked")
except TaskTimeoutError:
    print("Task exceeded 30s timeout")
except RateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after}s")