Bedsheet User Guide

1Getting Started

Installation

uv pip install bedsheet

Set Your API Key

Bedsheet uses Claude by default. Get your API key from console.anthropic.com:

export ANTHROPIC_API_KEY=your-key-here

Verify Installation

uvx bedsheet demo

This runs a demo showing multi-agent collaboration. If it works, you're ready!

2Your First Agent

Let's create the simplest possible agent - one that just responds to messages without any tools:

import asyncio
from bedsheet import Agent
from bedsheet.llm import AnthropicClient
from bedsheet.events import CompletionEvent

async def main():
    # Create an agent
    agent = Agent(
        name="Assistant",
        instruction="You are a helpful assistant. Be concise.",
        model_client=AnthropicClient(),
    )

    # Invoke the agent and get events
    async for event in agent.invoke(
        session_id="my-session",
        input_text="What is Python?"
    ):
        if isinstance(event, CompletionEvent):
            print(event.response)

asyncio.run(main())

Output

3Adding Tools

Agents become powerful when they can use tools. Let's give our agent the ability to check the weather:

import asyncio
from bedsheet import Agent, ActionGroup
from bedsheet.llm import AnthropicClient
from bedsheet.events import CompletionEvent, ToolCallEvent, ToolResultEvent

# Step 1: Create an ActionGroup to hold tools
tools = ActionGroup(name="WeatherTools")

# Step 2: Define a tool using the @action decorator
@tools.action(
    name="get_weather",
    description="Get the current weather for a city"
)
async def get_weather(city: str) -> dict:
    """In a real app, this would call a weather API."""
    weather_data = {
        "New York": {"temp": 72, "condition": "Sunny"},
        "London": {"temp": 58, "condition": "Cloudy"},
        "Tokyo": {"temp": 68, "condition": "Partly cloudy"},
    }
    return weather_data.get(city, {"temp": 70, "condition": "Unknown"})

async def main():
    # Step 3: Create an agent
    agent = Agent(
        name="WeatherBot",
        instruction="You help users check the weather. Use get_weather when asked.",
        model_client=AnthropicClient(),
    )

    # Step 4: Attach the tools to the agent
    agent.add_action_group(tools)

    # Step 5: Invoke
    async for event in agent.invoke("session-1", "What's the weather in Tokyo?"):
        if isinstance(event, ToolCallEvent):
            print(f"[Tool Call] {event.tool_name}({event.tool_input})")
        elif isinstance(event, ToolResultEvent):
            print(f"[Tool Result] {event.result}")
        elif isinstance(event, CompletionEvent):
            print(f"\n{event.response}")

asyncio.run(main())

Output

4Understanding Events

Bedsheet uses an event-driven architecture. Instead of returning a single response, invoke() yields events as things happen. This gives you full visibility into the agent's behavior.

All Event Types

from bedsheet.events import (
    # Single agent events
    TextTokenEvent,      # A token from streaming response
    ToolCallEvent,       # LLM wants to call a tool
    ToolResultEvent,     # Tool execution completed
    CompletionEvent,     # Agent finished with final response
    ErrorEvent,          # Something went wrong
    ThinkingEvent,       # Extended thinking content

    # Multi-agent events
    DelegationEvent,          # Supervisor delegating to agents
    CollaboratorStartEvent,   # A collaborator agent is starting
    CollaboratorEvent,        # Wraps events from collaborators
    CollaboratorCompleteEvent,# A collaborator finished
    RoutingEvent,             # Router picked an agent
)

Event Handling Pattern

async for event in agent.invoke(session_id, user_input):
    match event:
        case ToolCallEvent(tool_name=name, tool_input=args):
            print(f"Calling {name} with {args}")

        case ToolResultEvent(result=result, error=error):
            if error:
                print(f"Tool failed: {error}")
            else:
                print(f"Tool returned: {result}")

        case CompletionEvent(response=text):
            print(f"Final: {text}")

        case ErrorEvent(error=err):
            print(f"Error: {err}")

5Streaming Responses

By default, you get the complete response in CompletionEvent. But you can stream token-by-token for a better UX:

async for event in agent.invoke(session_id, user_input, stream=True):
    if isinstance(event, TextTokenEvent):
        # Print each word as it arrives
        print(event.token, end="", flush=True)
    elif isinstance(event, CompletionEvent):
        print()  # Newline after streaming completes

6Structured Outputs

Bedsheet supports Anthropic's native structured outputs feature. You define a schema, and Claude's constrained decoding ensures 100% compliance:

Option 1: Raw JSON Schema (No Dependencies)

from bedsheet.llm import AnthropicClient, OutputSchema

# Define your schema as a JSON Schema dict
schema = OutputSchema.from_dict({
    "type": "object",
    "properties": {
        "symbol": {"type": "string"},
        "recommendation": {
            "type": "string",
            "enum": ["buy", "sell", "hold"]
        },
        "confidence": {
            "type": "number",
            "minimum": 0,
            "maximum": 1
        },
        "reasoning": {"type": "string"}
    },
    "required": ["symbol", "recommendation", "confidence", "reasoning"]
})

# Use it with the LLM client
client = AnthropicClient()
response = await client.chat(
    messages=[{"role": "user", "content": "Analyze NVDA stock"}],
    system="You are a stock analyst.",
    output_schema=schema,
)

# Access the validated data - guaranteed to match your schema!
print(response.parsed_output)
# {"symbol": "NVDA", "recommendation": "buy", "confidence": 0.85, "reasoning": "..."}

Option 2: Pydantic Models (If You Prefer)

If you're already using Pydantic in your project, you can create schemas from your models:

from pydantic import BaseModel
from bedsheet.llm import OutputSchema

class StockAnalysis(BaseModel):
    symbol: str
    recommendation: str
    confidence: float
    reasoning: str

schema = OutputSchema.from_pydantic(StockAnalysis)

response = await client.chat(
    messages=[{"role": "user", "content": "Analyze AAPL"}],
    system="You are a stock analyst.",
    output_schema=schema,
)

# parsed_output is a dict matching your schema
print(response.parsed_output["recommendation"])  # "hold"

7Conversation Memory

Agents can remember previous messages in a session using Memory:

from bedsheet import Agent
from bedsheet.memory import InMemory

agent = Agent(
    name="Assistant",
    instruction="You are helpful.",
    model_client=AnthropicClient(),
    memory=InMemory(),  # Enable conversation memory
)

# First message
async for event in agent.invoke("session-1", "My name is Alice"):
    if isinstance(event, CompletionEvent):
        print(event.response)
# Output: "Nice to meet you, Alice!"

# Second message - agent remembers!
async for event in agent.invoke("session-1", "What's my name?"):
    if isinstance(event, CompletionEvent):
        print(event.response)
# Output: "Your name is Alice."

# Different session - no memory
async for event in agent.invoke("session-2", "What's my name?"):
    if isinstance(event, CompletionEvent):
        print(event.response)
# Output: "I don't know your name yet."

Memory Backends

from bedsheet.memory import InMemory, RedisMemory

# Development: Simple in-memory (lost when app restarts)
memory = InMemory()

# Production: Redis (persistent across restarts)
memory = RedisMemory(url="redis://localhost:6379")

8Multiple Tools

Agents can have many tools. The LLM decides which to use:

tools = ActionGroup(name="ResearchTools")

@tools.action(name="search_web", description="Search the web for information")
async def search_web(query: str) -> list:
    return [{"title": "Result 1", "snippet": "..."}]

@tools.action(name="get_page", description="Fetch content from a URL")
async def get_page(url: str) -> str:
    return "<html>...</html>"

@tools.action(name="summarize", description="Summarize long text")
async def summarize(text: str, max_words: int = 100) -> str:
    return "Summary..."

agent = Agent(
    name="Researcher",
    instruction="""You help users research topics.
    1. Search for relevant information
    2. Fetch pages if needed
    3. Summarize findings""",
    model_client=AnthropicClient(),
)
agent.add_action_group(tools)

9Your First Multi-Agent System

Now let's coordinate multiple agents. A Supervisor can delegate tasks to specialized agents:

from bedsheet import Agent, Supervisor, ActionGroup
from bedsheet.llm import AnthropicClient
from bedsheet.memory import InMemory
from bedsheet.events import (
    CompletionEvent, DelegationEvent,
    CollaboratorStartEvent, CollaboratorCompleteEvent
)

# === Agent 1: Translator ===
translator = Agent(
    name="Translator",
    instruction="You translate text. Respond with just the translation.",
    model_client=AnthropicClient(),
)

# === Agent 2: Summarizer ===
summarizer = Agent(
    name="Summarizer",
    instruction="You summarize text. Respond with just the summary.",
    model_client=AnthropicClient(),
)

# === Supervisor: Coordinator ===
coordinator = Supervisor(
    name="Coordinator",
    instruction="""You coordinate text processing.

    Collaborators:
    - Translator: For translation tasks
    - Summarizer: For summarization tasks

    Delegate to the appropriate agent based on the user's request.""",
    model_client=AnthropicClient(),
    memory=InMemory(),
    collaborators=[translator, summarizer],
    collaboration_mode="supervisor",
)

async def main():
    async for event in coordinator.invoke("s1", "Translate 'Hello' to Spanish"):
        if isinstance(event, DelegationEvent):
            agents = [d["agent_name"] for d in event.delegations]
            print(f"Delegating to: {agents}")

        elif isinstance(event, CollaboratorStartEvent):
            print(f"  [{event.agent_name}] Starting...")

        elif isinstance(event, CollaboratorCompleteEvent):
            print(f"  [{event.agent_name}] Done")

        elif isinstance(event, CompletionEvent):
            print(f"\nFinal: {event.response}")

Output

10Parallel Delegation

The real power: running multiple agents simultaneously.

# Supervisor with Parallel Delegation
advisor = Supervisor(
    name="InvestmentAdvisor",
    instruction="""You coordinate investment research.

    For stock analysis, delegate to BOTH agents IN PARALLEL:

    delegate(delegations=[
        {"agent_name": "MarketAnalyst", "task": "Analyze price data"},
        {"agent_name": "NewsResearcher", "task": "Find recent news"}
    ])

    Then synthesize their findings.""",
    model_client=AnthropicClient(),
    memory=InMemory(),
    collaborators=[market_analyst, news_researcher],
    collaboration_mode="supervisor",
)

0s  MarketAnalyst starts
0s  NewsResearcher starts  ← Both start!
2s  MarketAnalyst done
2s  NewsResearcher done
3s  Synthesis complete
─────────────────────────
Total: 3 seconds

0s  MarketAnalyst starts
2s  MarketAnalyst done
2s  NewsResearcher starts  ← Waits!
4s  NewsResearcher done
5s  Synthesis complete
─────────────────────────
Total: 5 seconds

11Deep Agent Hierarchies

Supervisors can contain other Supervisors, creating deep hierarchies:

# Level 3: Specialist agents
code_writer = Agent(name="CodeWriter", ...)
code_reviewer = Agent(name="CodeReviewer", ...)
test_writer = Agent(name="TestWriter", ...)

# Level 2: Team leads (Supervisors containing agents)
dev_lead = Supervisor(
    name="DevLead",
    instruction="Coordinate code writing and review.",
    collaborators=[code_writer, code_reviewer],
    ...
)

qa_lead = Supervisor(
    name="QALead",
    instruction="Coordinate testing efforts.",
    collaborators=[test_writer],
    ...
)

# Level 1: Project manager (Supervisor containing supervisors!)
project_manager = Supervisor(
    name="ProjectManager",
    instruction="""Manage software projects.
    - DevLead for implementation
    - QALead for testing""",
    collaborators=[dev_lead, qa_lead],  # Other supervisors!
    ...
)

12Router Mode

Sometimes you don't need synthesis - just route to the right specialist:

# Router just picks one and hands off
help_desk = Supervisor(
    name="HelpDesk",
    instruction="""Route programming questions to the right expert:
    - Python questions → PythonExpert
    - JavaScript questions → JavaScriptExpert
    - Database questions → DatabaseExpert

    Pick ONE expert that best matches.""",
    collaborators=[python_expert, javascript_expert, database_expert],
    collaboration_mode="router",  # Just route, don't synthesize
    ...
)

13Best Practices

Agent Design

1. Single Responsibility

Each agent should do one thing well:

code_writer = Agent(
    name="CodeWriter",
    instruction="Write code."
)
code_reviewer = Agent(
    name="CodeReviewer",
    instruction="Review code."
)

do_everything = Agent(
    name="DoEverything",
    instruction="Write, review,
    test, deploy, monitor..."
)

2. Clear Instructions

You summarize text.
- Under 100 words
- Focus on key points
- Use bullet points

Help with text stuff

3. Tool Descriptions Matter

The LLM uses descriptions to decide when to call tools:

@tools.action(
    name="get_weather",
    description="Get current weather
    for a city. Returns temperature
    in Fahrenheit and conditions."
)

@tools.action(
    name="get_weather",
    description="Weather stuff"
)

Multi-Agent Design

Error Handling

@tools.action(name="get_data", description="Fetch data")
async def get_data(id: str) -> dict:
    try:
        return await fetch(id)
    except NotFoundError:
        return {"error": f"No data found for id: {id}"}
    except Exception as e:
        return {"error": f"Failed to fetch: {str(e)}"}

Production Readiness

# Use Redis for production
from bedsheet.memory import RedisMemory
memory = RedisMemory(url="redis://your-redis:6379")

# Set iteration limits
supervisor = Supervisor(..., max_iterations=10)

# Log events for observability
async for event in agent.invoke(...):
    logger.info(f"Event: {event.type}", extra={"event": event})