Multi-agent AI,
without the boilerplate.

# Core library
pip install deepcrew-ai

# With OpenTelemetry support
pip install "deepcrew-ai[otel]"

# With dev dependencies (pytest, respx)
pip install "deepcrew-ai[dev]"

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GEMINI_API_KEY="AIza..."
export AWS_ACCESS_KEY_ID="..."      # for Bedrock
export AZURE_API_KEY="..."          # for Azure OpenAI

import asyncio
from deepcrew import Agent, run_agent, tool

@tool
def get_weather(city: str) -> dict:
    "Get current weather for a city."
    return {"city": city, "temperature": 22, "condition": "sunny"}

async def main():
    agent = Agent(
        name="assistant",
        model="openai/gpt-4o-mini",
        system_prompt="You are a helpful weather assistant.",
        tools=[get_weather],
    )
    result = await run_agent(
        agent,
        [{"role": "user", "content": "What's the weather in Tokyo?"}],
    )
    print(result.text)
    print(f"Tokens: {result.input_tokens}in / {result.output_tokens}out")

asyncio.run(main())

from deepcrew import Agent, RetryPolicy, FallbackChain, LoopConfig, InMemoryProvider
from deepcrew import WebSearchSkill
from deepcrew.mcp import HTTPMCP

agent = Agent(
    name="researcher",
    model="anthropic/claude-opus-4-8",
    system_prompt="You are a research specialist.",
    mcps=[HTTPMCP("https://my-mcp.example.com/mcp")],
    tools=[my_fn],
    skills=[WebSearchSkill()],              # v0.2.0: reusable skill bundles
    memory=InMemoryProvider(),              # v0.2.0: context persistence
    retry_policy=RetryPolicy(max_retries=3),# v0.2.0: retry on failure
    fallback_chain=FallbackChain([          # v0.2.0: model fallbacks
        "openai/gpt-4o",
        "gemini/gemini-2.0-flash",
    ]),
    loop_config=LoopConfig(max_iterations=3),# v0.2.0: outer refinement loop
    max_turns=10,
    temperature=0.7,
    max_tokens=4096,
)

from deepcrew import tool, Agent

@tool
def search_web(query: str, max_results: int = 5) -> list[str]:
    """Search the web and return URLs.

    Args:
        query (str): The search query.
        max_results (int): Maximum results to return.
    """
    return ["https://example.com/..."]

@tool(name="db_lookup", description="Query the internal database")
def lookup_database(table: str, id: int) -> dict:
    return {"id": id, "data": "..."}

@tool
async def fetch_url(url: str) -> str:
    "Fetch the HTML content of a URL."
    import httpx
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=10.0)
        return r.text[:3000]

agent = Agent("assistant", model="openai/gpt-4o",
              tools=[search_web, lookup_database, fetch_url])

from deepcrew import Agent, run_agent, ObservabilityConfig

agent = Agent("assistant", model="openai/gpt-4o")
obs   = ObservabilityConfig(otel_endpoint="http://localhost:4317")

result = await run_agent(
    agent,
    messages=[{"role": "user", "content": "Hello!"}],
    tool_defs=None,        # optional: pre-fetched ToolDef list
    queue=None,            # optional: asyncio.Queue for streaming
    agent_id=None,         # optional: override agent_id in events
    observability=obs,     # v0.2.0: optional OTel config
)

print(result.text)
print(result.confidence)      # v0.2.0: populated by APEX (None when direct)
print(result.loop_iterations) # v0.2.0: outer loop count

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from deepcrew import Agent, Orchestrator, ApexConfig

app = FastAPI()
orch = Orchestrator(
    agents=[Agent("assistant", model="openai/gpt-4o", system_prompt="Be helpful.")],
    apex_config=ApexConfig(cite_sources=True),
)

@app.post("/chat")
async def chat(query: str):
    async def event_stream():
        async for event in orch.stream(query):
            yield event.to_sse()   # "event: text_delta\ndata: {...}\n\n"
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

import asyncio
from deepcrew import Agent, run_agent
from deepcrew.types import EventType

queue = asyncio.Queue()
agent = Agent("bot", model="openai/gpt-4o")

task = asyncio.create_task(
    run_agent(agent, [{"role": "user", "content": "Hi"}], queue=queue)
)

while True:
    event = await queue.get()
    if event is None:
        break
    if event.event == EventType.TEXT_DELTA:
        print(event.data["chunk"], end="", flush=True)
    elif event.event == EventType.APEX_DONE:
        print(f"\nConfidence: {event.data['confidence']:.2f}")

await task

from deepcrew import Agent, WorkflowBuilder, ObservabilityConfig

obs = ObservabilityConfig(otel_endpoint="http://localhost:4317")

workflow = (
    WorkflowBuilder(observability=obs)   # v0.2.0: OTel tracing per step
    .add_agent("research", Agent("researcher", model="openai/gpt-4o-mini",
               system_prompt="Research the topic."), task="{input}")
    .add_agent("critique", Agent("critic", model="anthropic/claude-haiku-4-5-20251001",
               system_prompt="Find gaps."), task="Critique:\n{research}")
    .add_agent("expand",   Agent("expander", model="gemini/gemini-2.0-flash",
               system_prompt="Add technical depth."), task="Expand:\n{research}")
    .add_agent("report",   Agent("writer", model="openai/gpt-4o",
               system_prompt="Write a polished report."),
               task="Write a report.\n\nResearch:\n{research}\n\nCritique:\n{critique}\n\nDetails:\n{expand}")
    .then("research", "critique")
    .then("research", "expand")
    .then("critique", "report")
    .then("expand",   "report")
)

# research → [critique + expand in parallel] → report
result = await workflow.run("Future of renewable energy")
print(result.final_output.text)
print(f"Total tokens: {result.total_tokens}")

from deepcrew.types import EventType

async for event in workflow.stream("My topic"):
    if event.event == EventType.STEP_START:
        print(f"▶ {event.data['node']} starting...")
    elif event.event == EventType.TEXT_DELTA:
        print(event.data["chunk"], end="", flush=True)
    elif event.event == EventType.STEP_DONE:
        print(f"\n✓ {event.data['node']} done")
    elif event.event == EventType.DONE:
        print(f"\n\nFinal answer ready")

from deepcrew import Agent, Orchestrator, ApexConfig, tool

@tool
def search_web(query: str) -> str: ...
@tool
def run_code(code: str) -> str: ...

orch = Orchestrator(
    agents=[
        Agent("researcher", model="openai/gpt-4o-mini",
              system_prompt="Gather comprehensive facts."),
        Agent("analyst",    model="anthropic/claude-haiku-4-5-20251001",
              system_prompt="Evaluate trends and risks."),
        Agent("coder",      model="gemini/gemini-2.0-flash",
              system_prompt="Write and test code examples."),
    ],
    router_model="openai/gpt-4o-mini",
    apex_model="openai/gpt-4o",
    apex_config=ApexConfig(
        cite_sources=True,        # adds [source: agent_name] inline
        confidence_threshold=0.8,
        allow_tools=False,
    ),
    global_tools=[search_web, run_code],  # v0.2.0: router allocates per agent
    enable_spawn=True,                     # v0.2.0: agents can spawn sub-agents
    max_parallel_agents=5,
)

result = await orch.run("What are the key challenges of AI in healthcare?")
print(result.final_text)

# Streaming
async for event in orch.stream("..."):
    print(event.to_sse(), end="")

from deepcrew import Agent, run_agent
from deepcrew.mcp import StdioMCP

async with StdioMCP(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "."],
    env={"SOME_VAR": "value"},
) as mcp:
    agent = Agent("file_agent", model="openai/gpt-4o",
                  system_prompt="Help with file operations.", mcps=[mcp])
    result = await run_agent(agent, [{"role": "user", "content": "List Python files."}])

from deepcrew.mcp import HTTPMCP

async with HTTPMCP(
    url="https://my-mcp.example.com/mcp",
    headers={"Authorization": "Bearer sk-..."},
    timeout=30,
    retries=2,
) as mcp:
    tools = await mcp.list_tools()
    result = await mcp.call_tool("search", {"query": "hello world"})

from deepcrew.mcp import SSEMCP

async with SSEMCP("http://localhost:3000", headers={"X-API-Key": "secret"}) as mcp:
    tools = await mcp.list_tools()
    result = await mcp.call_tool("my_tool", {"arg": "value"})

from deepcrew import Agent, run_agent
from deepcrew.mcp import MCPManager, StdioMCP, HTTPMCP

async with MCPManager([
    StdioMCP("npx", ["-y", "@modelcontextprotocol/server-filesystem", "."]),
    HTTPMCP("https://search-mcp.example.com/mcp"),
    HTTPMCP("https://calendar-mcp.example.com/mcp"),
]) as manager:
    agent = Agent("super_agent", model="openai/gpt-4o",
                  system_prompt="Use all available tools.", mcps=[manager])
    result = await run_agent(agent, [{"role": "user", "content": "..."}])

async def run_agent(
    agent: Agent,
    messages: list[dict[str, Any]],
    *,
    tool_defs: list[ToolDef] | None = None,
    queue: asyncio.Queue[StreamEvent | None] | None = None,
    agent_id: str | None = None,
    observability: ObservabilityConfig | None = None,  # v0.2.0
) -> AgentResult

class Orchestrator:
    def __init__(
        self,
        agents: list[Agent],
        router_model: str = "openai/gpt-4o-mini",
        synthesizer_model: str | None = None,    # backward compat alias
        apex_model: str | None = None,            # v0.2.0
        apex_config: ApexConfig | None = None,    # v0.2.0
        router_system_prompt: str | None = None,
        synthesizer_system_prompt: str | None = None,
        max_parallel_agents: int = 5,
        global_tools: list[ToolDef] | None = None, # v0.2.0
        enable_spawn: bool = False,                # v0.2.0
    ) -> None

    async def run(self, query: str, context: dict | None = None) -> OrchestratorResult
    def stream(self, query: str, context: dict | None = None) -> AsyncGenerator[StreamEvent, None]

class WorkflowBuilder:
    def __init__(self, observability: ObservabilityConfig | None = None)  # v0.2.0
    def add_agent(self, name: str, agent: Agent, task: str | Callable = "{input}") -> WorkflowBuilder
    def then(self, predecessor: str, successor: str) -> WorkflowBuilder
    async def run(self, initial_input: str, context: dict | None = None) -> WorkflowResult
    def stream(self, initial_input: str, context: dict | None = None) -> AsyncGenerator[StreamEvent, None]

# APEX
class APEXSynthesizer:
    def __init__(self, model: str, config: ApexConfig | None = None)
    async def synthesize(self, query: str, results: list[AgentResult], ...) -> AgentResult

@dataclass class ApexConfig:
    confidence_threshold: float = 0.7
    cite_sources: bool = True
    allow_tools: bool = False
    system_prompt: str | None = None

# Loop
@dataclass class LoopConfig:
    max_iterations: int = 5
    convergence_fn: Callable[[AgentResult], bool] | None = None
    stop_condition: Callable[[AgentResult], bool] | None = None
    refine_prompt: str = "..."

async def run_agent_loop(agent, messages, *, tool_defs=None, queue=None, agent_id=None) -> AgentResult
async def search_loop(query, search_tool, *, agent, max_iterations=3, confidence_threshold=0.8, queue=None) -> AgentResult

# Memory
class MemoryProvider(ABC):  # store / retrieve / search / clear
class InMemoryProvider(MemoryProvider)
class FileMemoryProvider(MemoryProvider):
    def __init__(self, path: str | Path)

# Retry
@dataclass class RetryPolicy:
    max_retries: int = 3
    backoff_seconds: float = 1.0
    retry_on: tuple[type[Exception], ...] = (Exception,)
    exponential: bool = True

@dataclass class FallbackChain:
    models: list[str]

# Skills
class Skill(ABC):  # name / description / parameters / execute(**kwargs) -> str / to_tool_def()
def skill(fn=None, *, name=None, description=None)  # decorator
class SkillRegistry:  # register / get / list_all / clear
class WebSearchSkill(Skill)
class SummarizeSkill(Skill):
    def __init__(self, model: str = "openai/gpt-4o-mini")
class CodeExecutionSkill(Skill):
    def __init__(self, timeout: float = 10.0)

# Spawning
@dataclass class SpawnRequest:
    task: str
    tools: list[str] = []
    model: str | None = None
    system_prompt: str | None = None
    max_turns: int = 5

class ToolAllocator:
    def __init__(self, router_model: str)
    async def allocate(self, task: str, all_tools: list[ToolDef], max_tools: int = 10) -> list[ToolDef]

async def spawn_agent(request, all_tool_defs, parent_queue=None, router_model=..., parent_agent_id=...) -> AgentResult
def make_spawn_tool(all_tool_defs, parent_queue, router_model, parent_agent_id) -> ToolDef

# Observability
@dataclass class ObservabilityConfig:
    otel_endpoint: str | None = None
    service_name: str = "deepcrew"
    enabled: bool = True
    export_format: Literal["grpc","http"] = "grpc"

@dataclass class AgentResult:
    agent_id: str
    text: str
    tool_calls: list[dict]
    input_tokens: int
    output_tokens: int
    model: str
    confidence: float | None    # v0.2.0 — populated by APEX
    loop_iterations: int        # v0.2.0 — outer loop count
    total_tokens: int           # property

@dataclass class WorkflowResult:
    outputs: dict[str, AgentResult]
    final_output: AgentResult | None
    total_input_tokens: int
    total_output_tokens: int
    total_tokens: int  # property

@dataclass class OrchestratorResult:
    final_text: str
    agent_results: list[AgentResult]
    router_result: AgentResult | None
    total_input_tokens: int
    total_output_tokens: int
    total_tokens: int  # property