### Install documentation dependencies

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/BUILD_DOCS.md

Install the necessary packages for building documentation using the docs extra.

```bash
pip install -e ".[docs]"
```

--------------------------------

### Preview documentation locally

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/BUILD_DOCS.md

Start a local development server with live reloading to preview changes.

```bash
mkdocs serve
```

--------------------------------

### RedisStore Implementation Example

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/store.md

An example of how to create a custom RedisStore implementation by extending the abstract Store class.

```APIDOC
## RedisStore (Example Implementation)

### Description
This example demonstrates how to create a custom `RedisStore` by inheriting from the abstract `Store` class. This implementation uses the `redis-py` library to persist agent state in a Redis database.

### Method
- **__init__(self, host='localhost', port=6379, db=0)**: Initializes the Redis connection.

### Methods (Inherited from Store)
- **get(key: str) -> any**: Retrieves the value associated with the given key from Redis.
- **set(key: str, value: any)**: Sets the value for the given key in Redis.
- **has(key: str) -> bool**: Checks if a key exists in Redis.
- **items() -> dict**: Returns all key-value pairs from Redis. (Note: The implementation for fetching all items from Redis needs to be provided.)

### Request Example
```python
from agentswarm.datamodels import Store
import redis

class RedisStore(Store):
    def __init__(self, host='localhost', port=6379, db=0):
        self.r = redis.Redis(host=host, port=port, db=db)

    def get(self, key: str) -> any:
        return self.r.get(key)

    def set(self, key: str, value: any):
        self.r.set(key, value)

    def has(self, key: str) -> bool:
        return self.r.exists(key)
    
    def items(self) -> dict:
        # Implementation to fetch all keys...
        pass
```
```

--------------------------------

### Execute an Agent

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Demonstrates the basic setup and execution flow for a ResearchAgent using a local store and tracing.

```python
async def main():
    client = Client(api_key="YOUR_API_KEY")
    store = LocalStore()
    tracing = LocalTracing(trace_path="./traces")

    context = Context(
        trace_id="research-session-001",
        messages=[Message(type="user", content="Research AI trends from techcrunch.com")],
        store=store,
        tracing=tracing,
        default_llm=GeminiLLM(client=client)
    )

    agent = ResearchAgent(client)
    responses = await agent.execute("user-123", context)

    for msg in responses:
        print(f"[{msg.type}]: {msg.content}")

asyncio.run(main())
```

--------------------------------

### Reference-based Configuration

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/remote_protocol.md

Example of configuring a remote store using a named reference instead of sensitive connection strings.

```json
{
  "__class__": "agentswarm.datamodels.redis_store.RedisStore",
  "config": {
    "connection_ref": "PROD_REDIS"
  }
}
```

--------------------------------

### Implementing a Custom LLM Provider

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/index.md

Guidance on creating a custom LLM provider by subclassing the `LLM` abstract base class and implementing the `generate` method. Includes a Python code example.

```APIDOC
## Implementing a Custom LLM Provider

### Description
To integrate a new LLM provider (e.g., Anthropic, OpenAI, local Llama), you need to create a subclass of the `LLM` abstract base class and implement the `generate` asynchronous method. This method should handle the conversion of Agentswarm's `Message` and `LLMFunction` objects to the provider's specific format, call the provider's API, and parse the response into an `LLMOutput` object.

### Method
`async def generate(self, messages: list[Message], functions: list[LLMFunction] = None, feedback: FeedbackSystem = None) -> LLMOutput`

### Endpoint
N/A

### Parameters
- **messages** (list[Message]) - Required - A list of messages representing the conversation history.
- **functions** (list[LLMFunction], optional) - Optional - A list of function definitions (schemas) that the LLM can call.
- **feedback** (FeedbackSystem, optional) - Optional - A system for providing feedback, potentially for streaming responses.

### Request Body
N/A

### Request Example
```python
from agentswarm.datamodels import Message, LLMFunction
from agentswarm.llms import LLM, LLMOutput

class MyCustomLLM(LLM):
    def __init__(self, api_key: str, model_name: str = "gpt-4"):
        self.client = ... # Initialize your client
        self.model = model_name

    async def generate(self, messages: list[Message], functions: list[LLMFunction] = None, feedback: FeedbackSystem = None) -> LLMOutput:
        # 1. Convert Agentswarm Messages to provider format
        # 2. Convert LLMFunctions to provider tool schemas
        # 3. Call the API (optionally with streaming via feedback)
        # 4. Parse the response into LLMOutput (text + function_calls)
        pass
```

### Response
#### Success Response (200)
- **LLMOutput** (object) - An object containing the generated text and any function calls made by the LLM.

#### Response Example
N/A
```

--------------------------------

### Start Remote Execution Worker

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/examples/remote_execution/README.md

Command to initialize the worker HTTP server for handling remote agent requests.

```bash
PYTHONPATH=../../src:./ python3 worker.py
```

--------------------------------

### Define Custom ReActAgent with LLM and Tools

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/react.md

Subclass ReActAgent to specify the LLM and available tools for the agent. This is a common starting point for creating custom agents.

```python
class MyOrchestrator(ReActAgent):
    def get_llm(self, user_id: str):
        return GeminiLLM()
        
    def available_agents(self, user_id: str):
        return [
            SearchAgent(),
            DatabaseAgent(),
            # ...
        ]
```

--------------------------------

### Implement Custom Thinking Agent for ReActAgent

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/react.md

Override the `get_thinking_agent()` method to provide a custom implementation of the thinking process. This example shows how to emit user feedback during reasoning.

```python
class ResponsiveThinkingAgent(ThinkingAgent):
    async def execute(
        self, user_id: str, context: Context, input_args: ThinkingInput
    ) -> ThoughtResponse:
        context.emit_feedback(input_args.reasoning)
        return await super().execute(user_id, context, input_args)
```

```python
class MyOrchestrator(ReActAgent):
    def get_thinking_agent(self):
        return ResponsiveThinkingAgent()
```

--------------------------------

### GET /execute/status/{handler_id}

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/remote_protocol.md

Polls the status of an asynchronous execution request.

```APIDOC
## GET /execute/status/{handler_id}

### Description
Retrieves the current status of an asynchronous execution. If completed, it returns the final result.

### Method
GET

### Endpoint
/execute/status/{handler_id}

### Parameters
#### Path Parameters
- **handler_id** (string) - Required - The ID returned by the /execute/async endpoint.
```

--------------------------------

### Build the static site

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/BUILD_DOCS.md

Compile the markdown files into a static HTML site located in the site directory.

```bash
mkdocs build
```

--------------------------------

### Configure GeminiLLM Integration

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Initialize GeminiLLM with API keys or Vertex AI clients and define tool functions for agent use.

```python
from agentswarm.llms import GeminiLLM, LLMFunction
from agentswarm.datamodels import Message, LocalFeedbackSystem, Feedback
from google.genai import Client

# Initialize with API key or existing client
llm = GeminiLLM(api_key="YOUR_API_KEY", model="gemini-3-flash-preview")

# Or with Vertex AI
client = Client(
    vertexai=True,
    project="your-gcp-project",
    location="us-central1"
)
llm = GeminiLLM(client=client, model="gemini-3-flash-preview")

# Define function declarations for tool use
functions = [
    LLMFunction(
        name="search_web",
        description="Search the web for information",
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    )
]
```

--------------------------------

### Generate with Function Calling and Streaming

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Demonstrates generating responses using function calling and streaming output with feedback. Ensure the 'functions' object is defined elsewhere.

```python
from agentswarm.datamodels import Message

messages = [
    Message(type="system", content="You are a helpful assistant with web search."),
    Message(type="user", content="Find latest news about AI agents")
]

response = await llm.generate(messages, functions=functions)

print(f"Text: {response.text}")
print(f"Function calls: {response.function_calls}")
print(f"Tokens used: {response.usage.total_token_count}")
```

```python
from agentswarm.feedback import LocalFeedbackSystem

feedback = LocalFeedbackSystem()
feedback.subscribe(lambda fb: print(f"Chunk: {fb.payload}", end="", flush=True))

response = await llm.generate(messages, functions=functions, feedback=feedback)
```

--------------------------------

### Define and Execute a Type-Safe Agent System

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/README.md

Demonstrates defining input schemas, creating specialized agents, and orchestrating them with a ReActAgent. Requires Pydantic for schema definition and an LLM client for execution.

```python
from pydantic import BaseModel, Field
from agentswarm.agents import BaseAgent, ReActAgent
from agentswarm.datamodels import Context, Message

# 1. Define your Input schema
class ScraperInput(BaseModel):
    url: str = Field(description="The URL to scrape")

# 2. Create a specialized Agent
class ScraperAgent(BaseAgent[ScraperInput, str]):
    def id(self) -> str: return "scraper"
    
    async def execute(self, user_id: str, context: Context, input: ScraperInput) -> str:
        # Business logic: scrape the web...
        return f"Content of {input.url}: AgentSwarm is awesome!"

# 3. Create a Master Agent that coordinates others
class ResearchMaster(ReActAgent):
    def id(self) -> str: return "researcher"
    
    def prompt(self, user_id: str) -> str:
        return "You are a professional researcher. Use the scraper to gather info."

    def available_agents(self, user_id: str) -> list[BaseAgent]:
        return [ScraperAgent()]

# 4. Run it!
import asyncio
from agentswarm.datamodels import LocalStore, Message, Context
from agentswarm.llms import GeminiLLM
from google.genai import Client

async def main():
    # Setup dependencies
    client = Client(api_key="YOUR_API_KEY")
    llm = GeminiLLM(client=client, model="gemini-3-flash-preview")
    store = LocalStore()
    
    # Create the agent
    master = ResearchMaster()
    
    # Initialize Context with a user request
    context = Context(
        trace_id="unique-trace-id",
        messages=[Message(role="user", content="Visit appfactory.it and tell me what they do.")],
        store=store,
        default_llm=llm
    )
    
    # Execute!
    responses = await master.execute("user-123", context)
    
    for msg in responses:
        print(f"[{msg.role}]: {msg.content}")

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Execute Utility Agents in Python

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Demonstrates the use of GatheringAgent, TransformerAgent, and MergeAgent to process and retrieve data from a LocalStore.

```python
from agentswarm.agents import GatheringAgent, TransformerAgent, MergeAgent
from agentswarm.agents import GatheringAgentInput, TransformerAgentInput, MergeAgentInput
from agentswarm.datamodels import Context, LocalStore
from agentswarm.llms import GeminiLLM
from agentswarm.utils.tracing import LocalTracing

async def demo_utility_agents():
    store = LocalStore()
    tracing = LocalTracing()
    llm = GeminiLLM(api_key="YOUR_API_KEY")

    # Populate store with data
    store.set("raw_html", "<html><body>Article about AI...</body></html>")
    store.set("data_2023", "Revenue: $1M, Users: 10k")
    store.set("data_2024", "Revenue: $5M, Users: 50k")

    context = Context(
        trace_id="utils-demo",
        messages=[],
        store=store,
        tracing=tracing,
        default_llm=llm
    )

    # TransformerAgent: Apply LLM-based transformation
    transformer = TransformerAgent()
    result = await transformer.execute(
        "user-123",
        context,
        TransformerAgentInput(
            key="raw_html",
            cmd="Extract the main article text, removing all HTML tags"
        )
    )
    print(f"Transformed data stored at: {result.key}")

    # MergeAgent: Combine multiple data sources
    merger = MergeAgent()
    merged = await merger.execute(
        "user-123",
        context,
        MergeAgentInput(keys=["data_2023", "data_2024"])
    )
    print(f"Merged data stored at: {merged.key}")

    # GatheringAgent: Retrieve for display
    gatherer = GatheringAgent()
    output = await gatherer.execute(
        "user-123",
        context,
        GatheringAgentInput(key=merged.key)
    )
    print(f"Final output: {output.value}")
```

--------------------------------

### Manage Execution Context

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Shows how to initialize, copy, serialize, and merge execution contexts for sub-agent tasks and remote execution.

```python
from agentswarm.datamodels import Context, Message, LocalStore
from agentswarm.utils.tracing import LocalTracing
from agentswarm.llms import GeminiLLM
from google.genai import Client

# Initialize dependencies
client = Client(api_key="YOUR_API_KEY")
store = LocalStore()
tracing = LocalTracing(trace_path="./traces")
llm = GeminiLLM(client=client)

# Create initial context
context = Context(
    trace_id="main-trace-001",
    messages=[
        Message(type="system", content="You are a helpful assistant."),
        Message(type="user", content="Analyze this data set.")
    ],
    store=store,
    tracing=tracing,
    default_llm=llm,
    thoughts=[]  # Stores agent reasoning
)

# Copy context for isolated sub-agent execution (Tabula Rasa)
child_context = context.copy_for_execution()
# child_context has: clean messages=[], same store reference, new step_id

# Copy for iteration within a loop
iter_context = context.copy_for_iteration(
    step_id="iter_0",
    messages=context.messages
)

# Serialize/deserialize for remote execution
context_dict = context.to_dict()
restored_context = Context.from_dict(context_dict, default_llm=llm)

# Merge remote execution results back
context.merge(
    remote_context=restored_context,
    base_messages_count=2,  # Original message count
    base_thoughts_count=0,
    base_usage_count=0
)

# Debug output
print(context.debug_print())
```

--------------------------------

### Implementing a New Agent

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/base.md

Demonstrates how to create a custom agent by inheriting from BaseAgent and specifying input and output types using Pydantic models.

```APIDOC
## Implementing a New Agent

To create a custom agent, inherit from `BaseAgent` and specify your input/output types.

```python
from pydantic import BaseModel
from agentswarm.agents import BaseAgent
from agentswarm.datamodels import Context

class MyInput(BaseModel):
    query: str

class MyAgent(BaseAgent[MyInput, StrResponse]): 
    # Specifying [InputType, OutputType] generics is CRITICAL 
    # for automatic schema generation.

    def id(self) -> str:
        return "my_agent"

    def description(self, user_id: str) -> str:
        return "Helps the user with X"

    async def execute(self, user_id: str, context: Context, input: MyInput = None) -> StrResponse:
        # Your logic here
        return StrResponse(value=f"Processed {input.query}")
```
```

--------------------------------

### Implement a Custom Redis Store

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/store.md

Create a custom store implementation using Redis for data persistence. Requires the redis-python library.

```python
from agentswarm.datamodels import Store
import redis

class RedisStore(Store):
    def __init__(self, host='localhost', port=6379, db=0):
        self.r = redis.Redis(host=host, port=port, db=db)

    def get(self, key: str) -> any:
        return self.r.get(key)

    def set(self, key: str, value: any):
        self.r.set(key, value)

    def has(self, key: str) -> bool:
        return self.r.exists(key)
    
    def items(self) -> dict:
        # Implementation to fetch all keys...
        pass
```

--------------------------------

### Integrate MCP Servers with MCPBaseAgent

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Shows how to define custom MCP agents by subclassing MCPBaseAgent and executing tools discovered from an MCP server.

```python
import asyncio
from agentswarm.agents.mcp_agent import MCPBaseAgent
from agentswarm.datamodels import Context, LocalStore
from agentswarm.utils.tracing import LocalTracing
from mcp import StdioServerParameters

# Define connection to an MCP server
class SQLiteMCPAgent(MCPBaseAgent):
    def get_server_params(self) -> StdioServerParameters:
        return StdioServerParameters(
            command="uvx",
            args=["mcp-server-sqlite", "--db-path", "database.db"],
            env={}
        )

class GitMCPAgent(MCPBaseAgent):
    def get_server_params(self) -> StdioServerParameters:
        return StdioServerParameters(
            command="uvx",
            args=["mcp-server-git", "--repository", "."],
            env={}
        )

# Dummy tracing for examples
class NoOpTracing:
    def trace_agent(self, *args): pass
    def trace_loop_step(self, *args): pass
    def trace_agent_result(self, *args): pass
    def trace_agent_error(self, *args): pass

async def main():
    sqlite_factory = SQLiteMCPAgent()

    # Connect and discover tools
    async with sqlite_factory.connect() as mcp:
        agents = await mcp.get_agents()
        print(f"Discovered {len(agents)} MCP tools")

        for agent in agents:
            print(f"  - {agent.id()}: {agent.description('user')}")

        # Execute a tool
        context = Context(
            trace_id="mcp-demo",
            store=LocalStore(),
            messages=[],
            tracing=NoOpTracing(),
            thoughts=[]
        )

        # Find and use the query tool
        query_agent = next((a for a in agents if a.id() == "read_query"), None)
        if query_agent:
            result = await query_agent.execute(
                "user-123",
                context,
                {"query": "SELECT * FROM users LIMIT 10"}
            )
            print(f"Query result: {result}")

asyncio.run(main())
```

--------------------------------

### Initialize and use ReliableLLM

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/reliable_llm.md

Wraps a base LLM instance with retry and timeout logic. Ensure the base LLM is properly configured before wrapping.

```python
from agentswarm.llms import GeminiLLM, ReliableLLM

# 1. Create your base LLM
base_llm = GeminiLLM(api_key="your-api-key")

# 2. Wrap it with ReliableLLM
reliable_llm = ReliableLLM(
    llm=base_llm,
    timeout=10.0,       # 10 seconds timeout
    max_retries=5,      # retry up to 5 times
    retry_delay=2.0     # start with 2 seconds delay
)

# 3. Use it exactly like any other LLM
# response = await reliable_llm.generate(messages)
```

--------------------------------

### Implement a Console Tracer

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/tracing.md

Create a custom tracer by extending the Tracing class to output agent execution events to the terminal.

```python
from agentswarm.utils.tracing import Tracing
from agentswarm.datamodels import Context

class ConsoleTracing(Tracing):
    def trace_agent(self, context: Context, agent_id: str, arguments: dict):
        print(f"🚀 Starting Agent: {agent_id} with inputs: {arguments}")

    def trace_agent_result(self, context: Context, agent_id: str, result: any):
        print(f"✅ Agent {agent_id} completed: {result}")
    
    # ... implement other methods ...
```

--------------------------------

### LocalStore Implementation

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/store.md

The LocalStore provides an in-memory dictionary-based implementation of the Store interface, suitable for testing and single-instance applications.

```APIDOC
## LocalStore

### Description
The `LocalStore` is a concrete implementation of the `Store` interface that uses an in-memory Python dictionary for state persistence. It is ideal for development, testing, and applications where data persistence across restarts is not required.

### Methods
- **get(key: str) -> any**: Retrieves the value associated with the given key from the in-memory dictionary.
- **set(key: str, value: any)**: Sets the value for the given key in the in-memory dictionary.
- **has(key: str) -> bool**: Checks if a key exists in the in-memory dictionary.
- **items() -> dict**: Returns all key-value pairs currently stored in the in-memory dictionary.
```

--------------------------------

### POST /execute

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/remote_protocol.md

Executes an agent synchronously by transmitting the current context and input to the worker.

```APIDOC
## POST /execute

### Description
Executes an agent synchronously. The worker processes the request and returns the result along with the updated context.

### Method
POST

### Endpoint
/execute

### Request Body
- **user_id** (string) - Required - The ID of the user.
- **agent_id** (string) - Required - The ID of the agent to execute.
- **context** (object) - Required - The current execution context including trace_id, step_id, messages, and store configuration.
- **input** (object) - Required - The input data for the agent execution.

### Request Example
{
  "user_id": "user_123",
  "agent_id": "agent_abc",
  "context": {
    "trace_id": "t_1",
    "step_id": "s_1",
    "messages": [],
    "store": { "__class__": "path.to.Store", "config": {} }
  },
  "input": {}
}

### Response
#### Success Response (200)
- **result** (object) - The execution result.
- **updated_context** (object) - The delta for messages, thoughts, and usage.

#### Response Example
{
  "result": {},
  "updated_context": {}
}
```

--------------------------------

### Run Agent Tracing Script

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/examples/README.md

Execute the tracing script to visualize agent execution. Use 'last' for the latest trace or provide a specific trace ID.

```bash
python examples/tracing.py last|trace_id
```

--------------------------------

### Run Remote Execution Client

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/examples/remote_execution/README.md

Command to execute the client that proxies requests to the worker and synchronizes state.

```bash
PYTHONPATH=../../src:./ python3 client.py
```

--------------------------------

### POST /execute/async

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/remote_protocol.md

Initiates an asynchronous agent execution and returns a handler ID for polling.

```APIDOC
## POST /execute/async

### Description
Starts an asynchronous execution process. Returns a handler_id that can be used to check the status.

### Method
POST

### Endpoint
/execute/async
```

--------------------------------

### Standard Output Models

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/base.md

Details the standard output models provided by Agentswarm for consistent agent responses, including VoidResponse, StrResponse, KeyStoreResponse, ThoughtResponse, and CompletionResponse.

```APIDOC
## Standard Outputs

Agentswarm provides a set of predefined output models in `agentswarm.datamodels.responses`. Using these standard responses helps maintain consistency across your agents, especially when they are used as tools by a ReAct agent.

- **`VoidResponse`**: Use when the agent performs an action but returns no data (e.g., "Email sent").

- **`StrResponse`**: Use when the agent returns a simple string, that should be elaborated.

- **`KeyStoreResponse`**: Use when the agent stores a large object in the Store and returns only the key and a description. This prevents context pollution.

- **`ThoughtResponse`**: Used internally for reasoning steps.

- **`CompletionResponse`**: Use when the agent returns a completion. This type can be used to complete the reasoing loop of the ReAct agent. For example, if an agent generate a custom output (for example, an image or a custom html), that do not need to be further processed, shall use this respose.

### Example

```python
from agentswarm.datamodels import VoidResponse

class EmailAgent(BaseAgent[EmailInput, VoidResponse]):
    # ...
    async def execute(self, user_id, context, input):
        send_email(input)
        return VoidResponse()
```
```

--------------------------------

### Implement FeedbackSystem for real-time streaming

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Provides a publish-subscribe mechanism for streaming agent events. Useful for UI updates and progress monitoring.

```python
from agentswarm.datamodels import Feedback, LocalFeedbackSystem

# Create feedback system
feedback = LocalFeedbackSystem()

# Subscribe to events
def on_feedback(fb: Feedback):
    if fb.source == "llm":
        # Streaming LLM tokens
        print(fb.payload, end="", flush=True)
    elif fb.source == "agent":
        print(f"\n[Agent Event] {fb.payload}")

feedback.subscribe(on_feedback)

# Push events during execution
feedback.push(Feedback(source="agent", payload="Starting web scrape..."))
feedback.push(Feedback(source="llm", payload="Analyzing "))
feedback.push(Feedback(source="llm", payload="the results..."))

# Use with Context for automatic LLM streaming
context = Context(
    trace_id="streaming-demo",
    messages=[],
    store=LocalStore(),
    tracing=LocalTracing(),
    feedback=feedback  # Enable streaming
)

# Emit custom feedback from within agents
context.emit_feedback(payload="Processing complete", source="custom-agent")
```

--------------------------------

### Define structured agent response types

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Demonstrates various response classes for different agent output patterns, including store-based results, strings, and reasoning thoughts.

```python
from agentswarm.datamodels import (
    KeyStoreResponse,
    StrResponse,
    ThoughtResponse,
    VoidResponse,
    CompletionResponse
)
from agentswarm.agents import BaseAgent
from pydantic import BaseModel

# KeyStoreResponse - Result stored in shared Store
class DataFetcherAgent(BaseAgent[Input, KeyStoreResponse]):
    async def execute(self, user_id, context, input):
        context.store.set("result_key", fetched_data)
        return KeyStoreResponse(key="result_key", description="Fetched user data")

# StrResponse - Simple string to be elaborated by master agent
class CalculatorAgent(BaseAgent[Input, StrResponse]):
    async def execute(self, user_id, context, input):
        result = calculate(input.expression)
        return StrResponse(value=f"Result: {result}")

# ThoughtResponse - For reasoning/planning (used by ThinkingAgent)
class ThinkingAgent(BaseAgent[Input, ThoughtResponse]):
    async def execute(self, user_id, context, input):
        return ThoughtResponse(thought="I should first fetch data, then analyze it")
```

--------------------------------

### Implement a Custom LLM Provider

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/index.md

Create a subclass of LLM and implement the generate method to integrate new AI model providers.

```python
from agentswarm.datamodels import Message, LLMFunction
from agentswarm.llms import LLM, LLMOutput

class MyCustomLLM(LLM):
    def __init__(self, api_key: str, model_name: str = "gpt-4"):
        self.client = ... # Initialize your client
        self.model = model_name

    async def generate(self, messages: list[Message], functions: list[LLMFunction] = None, feedback: FeedbackSystem = None) -> LLMOutput:
        # 1. Convert Agentswarm Messages to provider format
        # 2. Convert LLMFunctions to provider tool schemas
        # 3. Call the API (optionally with streaming via feedback)
        # 4. Parse the response into LLMOutput (text + function_calls)
        pass
```

--------------------------------

### Set Gemini API Key

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/examples/README.md

Define your Gemini API key in a .env file for authentication.

```env
GEMINI_API_KEY=your_gemini_api_key
```

--------------------------------

### Store Interface

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/store.md

The abstract Store class defines the fundamental key-value interface for state persistence in Agentswarm.

```APIDOC
## Store Interface

### Description
The `Store` class is an abstract base class that defines the interface for state persistence in Agentswarm. Agents interact with the `Store` through a simple key-value API, allowing for flexible backend implementations.

### Methods
- **get(key: str) -> any**: Retrieves the value associated with the given key.
- **set(key: str, value: any)**: Sets the value for the given key.
- **has(key: str) -> bool**: Checks if a key exists in the store.
- **items() -> dict**: Returns all key-value pairs in the store. (Note: This method might require specific implementation details depending on the backend.)
```

--------------------------------

### Record execution traces with LocalTracing

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Captures agent calls, arguments, results, and errors into JSON files for debugging. Set TRACE_STORE_FULL environment variable to log full store content.

```python
from agentswarm.utils.tracing import LocalTracing
from agentswarm.datamodels import Context, Message, LocalStore
import os

# Create tracing system
tracing = LocalTracing(trace_path="./traces")

# Create context with tracing
store = LocalStore()
context = Context(
    trace_id="debug-session-001",
    messages=[Message(type="user", content="Analyze data")],
    store=store,
    tracing=tracing
)

# Trace agent execution manually (usually done by ReActAgent)
tracing.trace_agent(context, agent_id="scraper", arguments={"url": "https://example.com"})

# Trace results
tracing.trace_agent_result(context, agent_id="scraper", result={"key": "data_123"})

# Trace errors
try:
    raise ValueError("Network timeout")
except Exception as e:
    tracing.trace_agent_error(context, agent_id="scraper", error=e)

# Trace loop iterations
tracing.trace_loop_step(context, step_name="Iteration 3")

# Output is written to ./traces/debug-session-001.json as JSONL
# Enable full store content logging with environment variable:
os.environ["TRACE_STORE_FULL"] = "true"
```

--------------------------------

### MCPBaseAgent and MCPToolAgent Integration

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/mcp.md

Agentswarm's native support for the Model Context Protocol (MCP) allows agents to connect to external tools and data sources. MCPBaseAgent manages the connection, while MCPToolAgent wraps exposed tools.

```APIDOC
## MCP Agentswarm Integration

### Description
Agentswarm provides native support for the Model Context Protocol (MCP), enabling agents to connect to a wide range of external tools and data sources. This integration is managed through two key classes: `MCPBaseAgent` for handling the connection to an MCP server and `MCPToolAgent` for wrapping each tool exposed by the server into a standard `BaseAgent`.

### Classes

#### `agentswarm.agents.MCPBaseAgent`
Manages the connection to an MCP server, typically via stdio.

#### `agentswarm.agents.MCPToolAgent`
Automatically wraps each tool exposed by the MCP server into a standard `BaseAgent`.

### Example Usage

```python
from agentswarm.agents import MCPBaseAgent
from mcp import StdioServerParameters

class MyMCP(MCPBaseAgent):
    def get_server_params(self):
        return StdioServerParameters(command="uvx", args=["mcp-server-sqlite", ...])

async def main():
    factory = MyMCP()
    async with factory.connect() as agent_manager:
        tools = await agent_manager.get_agents()
        # tools is a list of MCPToolAgent instances you can now use!
```

### Request Body

This section is not applicable as the provided text does not describe any API endpoints with request bodies. The example focuses on agent initialization and connection.

### Response

This section is not applicable as the provided text does not describe any API endpoints with specific success or error responses. The example illustrates obtaining a list of `MCPToolAgent` instances.
```

--------------------------------

### LLM Streaming with Feedback

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/feedback.md

Provide a `FeedbackSystem` to `llm.generate` to stream response chunks. Subscribe to the system to receive and process these chunks.

```python
feedback_system = LocalFeedbackSystem()
feedback_system.subscribe(lambda fb: print(fb.payload, end="", flush=True))

await llm.generate(messages=messages, feedback=feedback_system)
```

--------------------------------

### Use VoidResponse for Agents with No Data Output

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/base.md

When an agent performs an action but does not return any specific data, use VoidResponse. This is useful for actions like sending an email.

```python
from agentswarm.datamodels import VoidResponse

class EmailAgent(BaseAgent[EmailInput, VoidResponse]):
    # ...
    async def execute(self, user_id, context, input):
        send_email(input)
        return VoidResponse()
```

--------------------------------

### Implement a Custom Agent with BaseAgent

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/base.md

Inherit from BaseAgent and define input/output types using Pydantic models. Specifying generics is critical for automatic schema generation. Implement the id, description, and execute methods for agent functionality.

```python
from pydantic import BaseModel
from agentswarm.agents import BaseAgent
from agentswarm.datamodels import Context

class MyInput(BaseModel):
    query: str

class MyAgent(BaseAgent[MyInput, StrResponse]): 
    # Specifying [InputType, OutputType] generics is CRITICAL 
    # for automatic schema generation.

    def id(self) -> str:
        return "my_agent"

    def description(self, user_id: str) -> str:
        return "Helps the user with X"

    async def execute(self, user_id: str, context: Context, input: MyInput = None) -> StrResponse:
        # Your logic here
        return StrResponse(value=f"Processed {input.query}")
```

--------------------------------

### Gemini LLM Implementation

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/index.md

Details on the `GeminiLLM` class, which provides integration with Google's Gemini models via Vertex AI or the Generative AI SDKs.

```APIDOC
## Gemini LLM Implementation

### Description
The `GeminiLLM` class is the Agentswarm implementation for interacting with Google's Gemini models. It connects to either Google Cloud's Vertex AI platform or the Google Generative AI SDKs, allowing users to leverage Gemini's capabilities within Agentswarm.

### Method
N/A (This is a specific implementation class)

### Endpoint
N/A

### Parameters
N/A

### Request Body
N/A

### Response
N/A
```

--------------------------------

### LLM Abstract Base Class

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/index.md

The LLM class serves as the core interface between Agentswarm and different AI model providers. It defines standard methods for text generation and tool handling, allowing for model-agnostic integration.

```APIDOC
## LLM Abstract Base Class

### Description
The `LLM` abstract base class defines a standard interface for interacting with Large Language Models (LLMs). It supports text generation from a list of messages and handling of tools, including defining function schemas and parsing tool calls from model responses.

### Method
Abstract Base Class

### Endpoint
N/A

### Parameters
N/A

### Request Body
N/A

### Response
N/A
```

--------------------------------

### ReliableLLM: Timeout and Retry Wrapper

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Wraps a base LLM with timeout protection and automatic retries using exponential backoff. Configure timeout, max retries, initial delay, and backoff factor.

```python
from agentswarm.llms import GeminiLLM, ReliableLLM
from agentswarm.datamodels import Message

# Create base LLM
base_llm = GeminiLLM(api_key="YOUR_API_KEY")

# Wrap with reliability features
reliable_llm = ReliableLLM(
    llm=base_llm,
    timeout=30.0,          # 30 second timeout per request
    max_retries=3,         # Retry up to 3 times
    retry_delay=1.0,       # Start with 1 second delay
    backoff_factor=2.0     # Double delay each retry (1s, 2s, 4s)
)

# Use like any other LLM - handles timeouts and retries automatically
messages = [Message(type="user", content="Explain quantum computing")]

try:
    response = await reliable_llm.generate(messages)
    print(response.text)
except Exception as e:
    print(f"Failed after all retries: {e}")
```

--------------------------------

### Implement Research Agent with ReActAgent

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Create a master orchestration agent that inherits from ReActAgent to coordinate research tasks. It defines LLM, prompt, and available agents, including custom ones like ScraperAgent.

```python
import asyncio
from typing import List
from agentswarm.agents import ReActAgent, BaseAgent, GatheringAgent, TransformerAgent, MergeAgent
from agentswarm.datamodels import Context, Message, LocalStore
from agentswarm.llms import LLM, GeminiLLM
from agentswarm.utils.tracing import LocalTracing
from google.genai import Client

# Define a custom scraper agent (from previous example)
class ScraperAgent(BaseAgent):
    def id(self) -> str:
        return "scraper"
    def description(self, user_id: str) -> str:
        return "Scrapes web content from URLs"
    async def execute(self, user_id, context, input):
        # Implementation here
        pass

# Create a master orchestration agent
class ResearchAgent(ReActAgent):
    def __init__(self, client: Client):
        super().__init__(max_iterations=100, max_concurrent_agents=5)
        self.client = client

    def get_llm(self, user_id: str) -> LLM:
        return GeminiLLM(client=self.client, model="gemini-3-flash-preview")

    def id(self) -> str:
        return "research-master"

    def description(self, user_id: str) -> str:
        return "Coordinates research tasks using multiple specialized agents"

    def prompt(self, user_id: str) -> str:
        return """You are a professional researcher. Use available tools to:
        1. Scrape relevant URLs for information
        2. Transform and filter data using natural language commands
        3. Merge multiple data sources into comprehensive reports
        Always execute independent tasks in parallel for efficiency."""

    def available_agents(self, user_id: str) -> List[BaseAgent]:
        # Combine default agents with custom ones
        return self.get_default_agents() + [ScraperAgent()]

```

--------------------------------

### Shared Key-Value Memory Store

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Utilize LocalStore to manage large data references between agents without bloating the conversation context.

```python
from agentswarm.datamodels import LocalStore, Store

# Create local in-memory store
store = LocalStore()

# Store data with keys
store.set("scraped_page_1", "<html>...large content...</html>")
store.set("processed_data", {"users": [...], "stats": {...}})

# Check if key exists
if store.has("scraped_page_1"):
    content = store.get("scraped_page_1")
    print(f"Retrieved content length: {len(content)}")

# Get all items (useful for debugging/tracing)
all_data = store.items()
print(f"Store contains {len(store)} items")

# Using Store in an agent
class DataProcessorAgent(BaseAgent):
    async def execute(self, user_id: str, context: Context, input):
        # Read from store
        raw_data = context.store.get(input.source_key)

        # Process and store result
        processed = self.process(raw_data)
        result_key = f"processed_{uuid.uuid4()}"
        context.store.set(result_key, processed)

        # Return only the key reference
        return KeyStoreResponse(key=result_key, description="Processed data")
```

--------------------------------

### Define Custom MCP Agent with Stdio Server

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/mcp.md

Subclass MCPBaseAgent to configure connection parameters for an MCP server, such as using stdio with a specific command and arguments. This is useful for establishing communication with MCP-compatible services.

```python
from agentswarm.agents import MCPBaseAgent
from mcp import StdioServerParameters

class MyMCP(MCPBaseAgent):
    def get_server_params(self):
        return StdioServerParameters(command="uvx", args=["mcp-server-sqlite", ...])

async def main():
    factory = MyMCP()
    async with factory.connect() as agent_manager:
        tools = await agent_manager.get_agents()
        # tools is a list of MCPToolAgent instances you can now use!
```

--------------------------------

### Define and Use an HttpRemoteAgent

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/remote.md

Instantiate an HttpRemoteAgent to act as a local proxy for a remote agent. Specify the base URL, the remote agent's ID, and the execution mode. This agent can then be used like any other agent to execute tasks remotely.

```python
from agentswarm.agents import HttpRemoteAgent, RemoteExecutionMode

# Define a proxy for a remote agent
remote_agent = HttpRemoteAgent(
    base_url="https://api.my-agent-swarm.com",
    remote_agent_id="data_analyst_agent",
    mode=RemoteExecutionMode.SYNC
)

# Use it like any other agent
result = await remote_agent.execute(user_id, context, input_data)
```

--------------------------------

### Implement CompletionResponse for final answers

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Use CompletionResponse to signal the termination of a ReActAgent loop and provide the final result.

```python
class FinalAnswerAgent(BaseAgent[Input, CompletionResponse]):
    async def execute(self, user_id, context, input):
        return CompletionResponse(value="The analysis is complete. Summary: ...")
```

--------------------------------

### Error Handling in Agents

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/agents/base.md

Explains Agentswarm's error handling strategy, where agents should raise exceptions for problems, allowing orchestrators like ReActAgent to catch and manage them.

```APIDOC
## Error Handling

Agentswarm follows standard Python idioms for error handling. **If an agent encounters a problem that prevents it from completing its task, it should raise an exception.**

The framework (specifically the `ReActAgent` or other orchestrators) is designed to catch these exceptions and handle them appropriately—typically by reporting the error back to the LLM so it can decide how to recover (e.g., by trying a different tool or modifying inputs).

```python
    async def execute(self, user_id, context, input):
        if not input.valid:
             # This will be caught by the calling agent and presented as an error
             raise ValueError(f"Invalid input: {input}")
        # ...
```
```

--------------------------------

### Emit Feedback from Context

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/feedback.md

Use `context.emit_feedback` to send feedback directly from an agent's execution context. Specify the payload and source.

```python
context.emit_feedback(payload="Starting task...", source="my_agent")
```

--------------------------------

### Define FeedbackSystem Interface

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/core/feedback.md

Abstract base class defining the protocol for feedback systems. Implementations must provide `push` and `subscribe` methods.

```python
class FeedbackSystem(ABC):
    @abstractmethod
    def push(self, feedback: Feedback):
        """Pushes a new feedback event."""
        pass

    @abstractmethod
    def subscribe(self, callback: Callable[[Feedback], None]):
        """Subscribes to feedback events."""
        pass
```

--------------------------------

### Create HTTP-based remote agent proxy

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Defines a remote agent that communicates via HTTP. Requires a base URL and execution mode configuration.

```python
class RemoteAnalyzerAgent(HttpRemoteAgent[AnalysisInput, KeyStoreResponse]):
    def __init__(self):
        super().__init__(
            base_url="https://worker.example.com",
            remote_agent_id="document-analyzer",
            mode=RemoteExecutionMode.SYNC,  # or ASYNC for polling
            timeout=60.0
        )

    def description(self, user_id: str) -> str:
        return "Analyzes documents using remote GPU-powered service"

# Use in a ReActAgent workflow
class MasterAgent(ReActAgent):
    def available_agents(self, user_id: str) -> list:
        return self.get_default_agents() + [RemoteAnalyzerAgent()]
```

--------------------------------

### Define Typed Web Scraper Agent

Source: https://context7.com/ai-agentswarm/agentswarm/llms.txt

Create a custom agent with strongly-typed input and output schemas using Pydantic models. The agent scrapes web pages and stores content in a shared store, returning a KeyStoreResponse.

```python
from pydantic import BaseModel, Field
from agentswarm.agents import BaseAgent
from agentswarm.datamodels import Context, KeyStoreResponse
import uuid

# Define strongly-typed input schema
class WebScraperInput(BaseModel):
    url: str = Field(description="The URL to scrape")
    extract_links: bool = Field(default=False, description="Whether to extract links")

# Create a custom agent with typed input/output
class WebScraperAgent(BaseAgent[WebScraperInput, KeyStoreResponse]):
    def id(self) -> str:
        return "web-scraper"

    def description(self, user_id: str) -> str:
        return "Scrapes web pages and stores content in the shared store."

    async def execute(
        self, user_id: str,
        context: Context,
        input: WebScraperInput
    ) -> KeyStoreResponse:
        # Business logic: fetch and process the URL
        content = f"Scraped content from {input.url}"

        # Store result in shared store
        key = f"scraper_{uuid.uuid4()}"
        context.store.set(key, content)

        return KeyStoreResponse(
            key=key,
            description=f"Scraped content from {input.url}"
        )

# Get auto-generated JSON schema for LLM function calls
agent = WebScraperAgent()
print(agent.input_parameters())
# Output: {'type': 'object', 'properties': {'url': {'type': 'string', 'description': 'The URL to scrape'}, ...}}
```

--------------------------------

### Reliable LLM Wrapper

Source: https://github.com/ai-agentswarm/agentswarm/blob/master/docs/llms/index.md

Information about the `ReliableLLM` wrapper, which enhances any LLM with retry and timeout capabilities.

```APIDOC
## Reliable LLM (Wrapper)

### Description
The `ReliableLLM` is a wrapper class designed to add robustness to any existing LLM implementation. It automatically incorporates retry logic and timeout mechanisms, ensuring more dependable interactions with LLMs, especially in scenarios prone to network issues or intermittent failures.

### Method
Wrapper Class

### Endpoint
N/A

### Parameters
N/A

### Request Body
N/A

### Response
N/A

### Further Information
[Learn more about Reliable LLM](./reliable_llm.md)
```