Examples

from agentspan.agents import Agent, run

agent = Agent(name="greeter", model="openai/gpt-4o")
result = run(agent, "Say hello and tell me a fun fact about Python.")
print(result.output['result'])

from agentspan.agents import Agent, http_tool, run

weather = http_tool(
    name="get_weather",
    description="Get current weather",
    url="https://api.open-meteo.com/v1/forecast",
    method="GET",
    input_schema={...},
)

agent = Agent(name="weather_bot", model="openai/gpt-4o", tools=[weather])

from agentspan.agents import Agent, run

researcher = Agent(name="researcher", model="openai/gpt-4o",
                   instructions="Research the topic.")
writer = Agent(name="writer", model="openai/gpt-4o",
               instructions="Write an article.")
editor = Agent(name="editor", model="openai/gpt-4o",
               instructions="Polish the article.")

result = run(researcher >> writer >> editor, "Durable AI agents")

@tool(approval_required=True)
def transfer_funds(from_acct: str, to_acct: str, amount: float) -> dict:
    """Transfer funds. Requires human approval."""
    return process_transfer(from_acct, to_acct, amount)

handle = start(agent, "Transfer $5000 from checking to savings")
for event in handle.stream():
    if event.type == EventType.WAITING:
        handle.approve()  # or handle.reject("Too risky")

from agentspan.agents import RegexGuardrail, Guardrail, GuardrailResult

no_pii = RegexGuardrail(
    patterns=[r"\b\d{3}-\d{2}-\d{4}\b"],
    name="no_ssn",
    message="Do not include SSNs.",
)

def word_limit(content: str) -> GuardrailResult:
    if len(content.split()) > 200:
        return GuardrailResult(passed=False, message="Too long.")
    return GuardrailResult(passed=True)

agent = Agent(..., guardrails=[no_pii, Guardrail(word_limit, on_fail="retry")])

from agentspan.agents import Agent, run, SemanticMemory

memory = SemanticMemory(namespace="user:alice", top_k=3)
memory.store("Alice is on the Enterprise plan since March 2021.")
memory.store("Alice reported a billing issue on invoice #1042.")

agent = Agent(name="support", model="openai/gpt-4o", memory=memory)
result = run(agent, "What do we know about Alice's account?")

from agentspan.agents.code_executor import DockerCodeExecutor

executor = DockerCodeExecutor(image="python:3.12-slim", timeout=30)

agent = Agent(
    name="data_analyst",
    model="openai/gpt-4o",
    tools=[executor.as_tool()],
    instructions="Write and execute Python to answer data questions.",
)
result = run(agent, "Calculate the first 20 Fibonacci numbers.")

from agentspan.agents import stream, EventType

for event in stream(agent, "Write about distributed systems"):
    if event.type == EventType.TOOL_CALL:
        print(f"  -> {event.tool_name}({event.input})")
    elif event.type == EventType.DONE:
        print(event.output)

from agentspan.agents.testing import mock_run, MockEvent, expect

result = mock_run(agent, "Capital of France?", events=[
    MockEvent.tool_call("search_web", {"query": "capital of France"}),
    MockEvent.tool_result("search_web", "Paris is the capital of France."),
    MockEvent.done("The capital of France is Paris."),
])

expect(result).completed().output_contains("Paris").used_tool("search_web")

from agentspan.agents import start, AgentHandle

# Launch many agents concurrently
handles: list[AgentHandle] = [
    start(agent, f"Analyze customer {id}")
    for id in customer_ids
]

# Poll until all complete
results = [h.stream().get_result() for h in handles]