---
name: elsai-agent-sdk
description: Complete guide for building AI agents with the Elsai Agent SDK — covers installation, tools, models, hooks, sessions, and multi-agent patterns
allowed-tools: Read Write Edit Bash
metadata:
  version: "0.1.0"
  install: "pip install --extra-index-url https://elsai-agents.elsai.ai/root/ elsai-agents==0.1.0"
---

# Elsai Agent SDK — Developer Skill

Use this skill whenever the developer asks about building, configuring, or debugging agents with the Elsai Agent SDK.

---

## Installation

```bash
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ elsai-agents==0.1.0

# With model provider extras
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[anthropic]==0.1.0"
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[openai]==0.1.0"
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[gemini]==0.1.0"
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[litellm]==0.1.0"
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[ollama]==0.1.0"
```

---

## 1. Building an Agent

### Minimal agent (defaults to Amazon Bedrock)
```python
from elsai import Agent

agent = Agent()
result = agent("Explain quantum computing in one paragraph")
print(result)
```

### Agent with model, system prompt, and tools
```python
from elsai import Agent, tool
from elsai.models.anthropic import AnthropicModel

@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get current weather for a city.

    Args:
        city: Name of the city.
        unit: Temperature unit — 'celsius' or 'fahrenheit'.
    """
    return f"Weather in {city}: 22°C, sunny"

agent = Agent(
    model=AnthropicModel(model_id="claude-sonnet-4-6"),
    system_prompt="You are a helpful travel assistant.",
    tools=[get_weather],
    name="TravelBot",
    description="Helps with travel planning",
)
result = agent("What is the weather in Paris?")
```

### Available model providers
| Provider | Import | model_id example |
|---|---|---|
| Amazon Bedrock (default) | `from elsai.models import BedrockModel` | `"us.amazon.nova-pro-v1:0"` |
| Anthropic | `from elsai.models.anthropic import AnthropicModel` | `"claude-sonnet-4-6"` |
| OpenAI | `from elsai.models.openai import OpenAIModel` | `"gpt-4o"` |
| Google Gemini | `from elsai.models.gemini import GeminiModel` | `"gemini-2.0-flash"` |
| Ollama (local) | `from elsai.models.ollama import OllamaModel` | `"llama3.2"` |
| LiteLLM | `from elsai.models.litellm import LiteLLMModel` | `"anthropic/claude-3-5-sonnet"` |

---

## 2. Creating Tools

Rules:
- Docstring is **required** — the model reads it to decide when to call the tool
- All parameters must have type annotations
- Return a string, dict, Pydantic model, or any JSON-serialisable value

```python
from elsai import tool
from elsai.types.tools import ToolContext

# Basic tool
@tool
def calculate(expression: str) -> str:
    """Evaluate a Python math expression.

    Args:
        expression: Valid Python math expression e.g. '2 ** 10'.
    """
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

# Custom name and description
@tool(name="search_web", description="Search the internet for current information")
def search(query: str, max_results: int = 5) -> str:
    """Search the web."""
    ...

# Async tool
@tool
async def fetch_url(url: str) -> str:
    """Fetch the content of a URL asynchronously."""
    import aiohttp
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            return await r.text()

# Streaming tool (yields intermediate events, returns final result)
@tool
async def long_task(steps: int) -> str:
    """Run a multi-step background task, reporting progress.

    Args:
        steps: Number of steps to execute.
    """
    for i in range(steps):
        yield f"Step {i+1}/{steps} in progress..."
    return "All steps completed."

# Tool with access to agent context
@tool(context=True)
def stateful_tool(query: str, tool_context: ToolContext) -> str:
    """Tool that reads agent state.

    Args:
        query: The query to process.
    """
    agent = tool_context["agent"]
    state = agent.state.get()
    return f"State={state}, query={query}"

# Direct tool invocation (bypass LLM)
agent = Agent(tools=[calculate])
result = agent.tool.calculate(expression="42 * 7")
```

### MCP tools
```python
from mcp import StdioServerParameters, stdio_client
from mcp.client.sse import sse_client
from elsai.tools.mcp.mcp_client import MCPClient

# stdio transport (command-line MCP servers)
mcp = MCPClient(
    lambda: stdio_client(StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "."],
    ))
)

# SSE transport (HTTP MCP servers)
mcp = MCPClient(lambda: sse_client("http://localhost:8000/sse"))

# Use as a context manager — handles startup/shutdown automatically
with mcp:
    tools = mcp.list_tools_sync()
    agent = Agent(tools=tools)
    result = agent("List all Python files here")
```

---

## 3. Hooks — Lifecycle Events

Use hooks to observe, modify, or intercept the agent at any lifecycle point.

```python
from elsai import Agent
from elsai.hooks import (
    BeforeModelCallEvent,
    AfterModelCallEvent,
    BeforeToolCallEvent,
    AfterToolCallEvent,
    BeforeInvocationEvent,
    AfterInvocationEvent,
    AgentInitializedEvent,
    AgentErrorEvent,
    MessageAddedEvent,
)

# Log every model call
def log_model(event: BeforeModelCallEvent) -> None:
    print(f"[model] agent={event.agent.name}")

# Block a specific tool
def guard(event: BeforeToolCallEvent) -> None:
    if event.tool_use["name"] == "dangerous_op":
        event.cancel_tool = "Blocked by policy"

# Auto-retry a tool once on failure
def retry_once(event: AfterToolCallEvent) -> None:
    if event.exception and not event.retry:
        event.retry = True

# Catch errors
def on_error(event: AgentErrorEvent) -> None:
    print(f"[error] {event.exception}")

agent = Agent(hooks=[log_model, guard, retry_once, on_error])

# Or add after construction
agent.add_hook(log_model)
```

### Reusable HookProvider
```python
from elsai.hooks.registry import HookProvider, HookRegistry

class AuditLogger(HookProvider):
    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
        registry.add_callback(BeforeInvocationEvent, self._on_start)
        registry.add_callback(AfterInvocationEvent,  self._on_end)

    def _on_start(self, event: BeforeInvocationEvent) -> None:
        print(f"[audit] start agent={event.agent.name}")

    def _on_end(self, event: AfterInvocationEvent) -> None:
        print(f"[audit] end stop_reason={event.result.stop_reason if event.result else 'n/a'}")

agent = Agent(hooks=[AuditLogger()])
```

---

## 4. Sessions — Persistent Conversations

```python
from elsai import Agent
from elsai.session.file_session_manager import FileSessionManager
from elsai.session.s3_session_manager import S3SessionManager

# File-based (local)
# session_id groups all agents in one session; agent_id identifies the agent within that session
session = FileSessionManager(session_id="chat-42", storage_dir="./sessions")
agent = Agent(
    agent_id="assistant",         # agent key within the session — must be stable
    session_manager=session,
)
agent("My name is Alice and I like Python")
# Restart process — conversation is automatically restored
agent("What do you know about me?")  # remembers Alice + Python

# S3-based (production)
session = S3SessionManager(
    session_id="chat-42",
    bucket="my-sessions-bucket",  # param is 'bucket', not 'bucket_name'
    prefix="agents/",             # param is 'prefix', not 's3_prefix'
)
agent = Agent(agent_id="assistant", session_manager=session)
```

**Rules:** `session_id` is required on the manager; `agent_id` on the Agent identifies the agent within that session.

---

## 5. Multi-Agent Patterns

### Agent-as-Tool
```python
researcher = Agent(
    name="researcher",
    description="Searches and retrieves information on any topic",
    tools=[web_search_tool],
)
writer = Agent(
    name="writer",
    tools=[researcher.as_tool()],
)
result = writer("Write a blog post about AI agents")
```

### Graph — deterministic pipeline
```python
from elsai.multiagent.graph import GraphBuilder

planner  = Agent(name="planner",  system_prompt="Break the task into clear steps.")
executor = Agent(name="executor", system_prompt="Execute each step methodically.")
reviewer = Agent(name="reviewer", system_prompt="Review and summarise the output.")

builder = GraphBuilder()
builder.add_node(planner,  "plan")
builder.add_node(executor, "exec")
builder.add_node(reviewer, "review")
builder.add_edge("plan",  "exec")
builder.add_edge("exec",  "review")
builder.set_entry_point("plan")

# Optional conditional edge
builder.add_edge("exec", "review", condition=lambda state: bool(state.results.get("exec")))

graph = builder.build()
result = graph("Build a REST API for a todo app")
# result.status, result.completed_nodes, result.results["review"].message
```

### Swarm — autonomous handoffs
```python
from elsai.multiagent.swarm import Swarm

researcher = Agent(
    name="researcher",
    system_prompt="Research topics. Hand off calculations to the analyst.",
    tools=[web_search],
)
analyst = Agent(
    name="analyst",
    system_prompt="Analyse data. Hand off research to the researcher.",
    tools=[calculate],
)

swarm = Swarm(agents=[researcher, analyst], entry_agent=researcher)
result = swarm("Analyse AI market growth from 2024 to 2030")
```

---

## 6. Structured Output

```python
from pydantic import BaseModel
from elsai import Agent

class Report(BaseModel):
    title: str
    key_points: list[str]
    recommendation: str

agent = Agent()
result = agent(
    "Analyse Python vs JavaScript for backend development",
    structured_output_model=Report,
)
report: Report = result.structured_output
print(report.title, report.recommendation)
```

---

## 7. Streaming

```python
import asyncio
from elsai import Agent

async def stream():
    agent = Agent()
    async for event in agent.stream_async("Write a short story"):
        if "data" in event:
            print(event["data"], end="", flush=True)

asyncio.run(stream())
```

---

## 8. Conversation Management

```python
from elsai.agent.conversation_manager import SlidingWindowConversationManager
from elsai.agent.conversation_manager import SummarizingConversationManager

# Keep last N messages — param is 'window_size', default 40
agent = Agent(conversation_manager=SlidingWindowConversationManager(window_size=20))

# Summarise old messages instead of dropping them
# summary_ratio: fraction of oldest messages to summarise (default 0.3)
# preserve_recent_messages: minimum recent messages to always keep (default 10)
agent = Agent(
    conversation_manager=SummarizingConversationManager(
        summary_ratio=0.4,
        preserve_recent_messages=10,
    )
)
```

---

## 9. Agent State

```python
agent = Agent(state={"user_id": "u123", "session_count": 0})

# Read
agent.state.get()                   # full dict
agent.state.get("user_id")          # single key

# Write
agent.state["session_count"] += 1
agent.state.update({"last_seen": "2026-05-26"})
```

---

## 10. AgentResult Reference

```python
result = agent("Hello!")
str(result)                     # final text
result.stop_reason              # "end_turn" | "max_tokens" | "cancelled"
result.message                  # raw assistant message dict
result.metrics.total_tokens     # token usage
result.structured_output        # Pydantic model (when structured_output_model was set)
```

---

## Common Mistakes to Avoid

1. **Missing docstring on `@tool`** — the model won't know when to call it.
2. **Forgetting `agent.cleanup()`** — leaks MCP server processes.
3. **Session without `agent_id`** — conversation won't be keyed correctly.
4. **Swarm handoff not in system prompt** — agents won't know when to delegate.
5. **Graph entry point not set** — `builder.set_entry_point("node_id")` is required.
6. **Hook returns a value** — hooks must mutate the event object, not return values.
7. **`agent_id` with path separators** — use plain strings like `"user-123"`, not `"user/123"`.