# Elsai Agent SDK — Claude Code Skill

This file teaches Claude Code how to help developers build AI agents with the **Elsai Agent SDK**.

## Installation

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

Optional provider extras:
```bash
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ \
  "elsai-agents[anthropic]==0.1.0"   # Anthropic Claude
  "elsai-agents[openai]==0.1.0"      # OpenAI / Azure
  "elsai-agents[gemini]==0.1.0"      # Google Gemini
  "elsai-agents[ollama]==0.1.0"      # Ollama local models
  "elsai-agents[litellm]==0.1.0"     # LiteLLM (100+ providers)
```

## Core Concepts

### 1. Minimal Agent

```python
from elsai import Agent

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

`Agent()` defaults to **Amazon Bedrock** (uses `~/.aws` credentials or env vars).

### 2. Agent with a Specific Model

```python
from elsai import Agent
from elsai.models import BedrockModel
from elsai.models.anthropic import AnthropicModel
from elsai.models.openai import OpenAIModel
from elsai.models.gemini import GeminiModel
from elsai.models.ollama import OllamaModel
from elsai.models.litellm import LiteLLMModel

# Amazon Bedrock (default) — model ID string shortcut
agent = Agent(model="us.amazon.nova-pro-v1:0")

# Anthropic
agent = Agent(model=AnthropicModel(model_id="claude-sonnet-4-6"))

# OpenAI
agent = Agent(model=OpenAIModel(model_id="gpt-4o"))

# Google Gemini
agent = Agent(model=GeminiModel(model_id="gemini-2.0-flash"))

# Ollama (local)
agent = Agent(model=OllamaModel(model_id="llama3.2"))

# LiteLLM (any provider)
agent = Agent(model=LiteLLMModel(model_id="anthropic/claude-3-5-sonnet"))
```

### 3. System Prompt

```python
agent = Agent(
    model="us.amazon.nova-pro-v1:0",
    system_prompt="You are a helpful Python coding assistant. Always include type hints.",
)
```

---

## Tools

### @tool Decorator (simplest)

```python
from elsai import Agent, tool

@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'.
    """
    # real implementation here
    return f"Weather in {city}: 22°{unit[0].upper()}, sunny"

agent = Agent(tools=[get_weather])
result = agent("What's the weather in Paris?")
```

**Rules for @tool functions:**
- Docstring description is required — it's sent to the model
- Google-style Args section documents parameters
- Type hints are required for all params (used to build the JSON schema)
- Return `str`, `dict`, or any JSON-serialisable value
- For structured errors: `return {"status": "error", "content": [{"text": "..."}]}`

### @tool with custom name / description

```python
@tool(name="search_web", description="Search the internet for up-to-date information")
def search(query: str, max_results: int = 5) -> str:
    """Search the web."""
    ...
```

### Async tool

```python
@tool
async def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()
```

### Streaming tool (yields intermediate events)

```python
@tool
async def long_task(steps: int) -> str:
    """Run a long multi-step task."""
    for i in range(steps):
        yield f"Step {i+1}/{steps} complete..."  # intermediate events
    return "All steps done."                      # final result
```

### ToolContext — access agent & invocation state

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

@tool(context=True)
def stateful_tool(query: str, tool_context: ToolContext) -> str:
    """Tool that inspects agent state."""
    agent = tool_context["agent"]
    state = agent.state.get()
    return f"Agent state: {state}, query: {query}"
```

### Direct tool call (bypass LLM)

```python
result = agent.tool.get_weather(city="Tokyo")
```

### Load tools from directory (hot-reload)

```python
agent = Agent(load_tools_from_directory=True)
# Loads & watches ./tools/*.py — any @tool function is auto-registered
```

### 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", "."],
    ))
)

# 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 in this directory")
```

---

## Structured Output

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

class WeatherReport(BaseModel):
    city: str
    temperature: float
    conditions: str
    forecast: list[str]

agent = Agent()
result = agent(
    "Give me a weather report for London",
    structured_output_model=WeatherReport,
)
report: WeatherReport = result.structured_output
print(report.city, report.temperature)
```

---

## Streaming

### Async streaming

```python
import asyncio
from elsai import Agent

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

asyncio.run(main())
```

---

## Hooks (Lifecycle Events)

Hooks let you observe and modify the agent at every stage.

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

agent = Agent()

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

agent.add_hook(on_before_model)

# Intercept and cancel a tool
def guard_tool(event: BeforeToolCallEvent) -> None:
    if event.tool_use["name"] == "dangerous_tool":
        event.cancel_tool = "Tool blocked by policy"

agent.add_hook(guard_tool)

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

agent.add_hook(retry_on_error)
```

### Hook at construction time

```python
agent = Agent(
    hooks=[on_before_model, guard_tool],
)
```

### Custom HookProvider (reusable across agents)

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

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

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

    def _log_end(self, event: AfterInvocationEvent) -> None:
        print(f"[audit] end result={event.result}")

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

---

## State

Agent state persists across invocations and is accessible from tools.

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

# Read
agent.state.get()           # {"user_id": "u123", "preferences": {}}

# Write
agent.state["cart"] = ["item1"]
agent.state.update({"preferences": {"theme": "dark"}})
```

---

## Sessions (Persistence)

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

# session_id groups the session; agent_id identifies the agent within it
session = FileSessionManager(session_id="chat-alice", storage_dir="./sessions")

agent = Agent(
    agent_id="assistant",         # agent key within the session
    session_manager=session,
)

agent("Hello, remember my name is Alice")
# Restart the process — conversation is restored from disk
agent("What is my name?")         # → "Your name is Alice"
```

### S3 Session

```python
from elsai.session.s3_session_manager import S3SessionManager

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

---

## Multi-Agent Patterns

### Agent-as-Tool

```python
from elsai import Agent

researcher = Agent(
    name="researcher",
    description="Searches and retrieves information on any topic",
    tools=[web_search_tool],
)

writer = Agent(
    name="writer",
    tools=[researcher.as_tool()],  # researcher becomes a callable tool
)

result = writer("Write a blog post about quantum computing")
```

### Graph (deterministic pipeline)

```python
from elsai import Agent, tool
from elsai.multiagent.graph import GraphBuilder

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

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")

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

#### Conditional edges

```python
def needs_review(state) -> bool:
    return state.results.get("exec") is not None

builder.add_edge("exec", "review", condition=needs_review)
```

### Swarm (autonomous handoffs)

```python
from elsai.multiagent.swarm import Swarm

researcher = Agent(
    name="researcher",
    system_prompt=(
        "You research topics. When you need analysis, hand off to the analyst."
    ),
    tools=[web_search_tool],
)
analyst = Agent(
    name="analyst",
    system_prompt=(
        "You analyse data. When you need research, hand off to the researcher."
    ),
    tools=[calculate_tool],
)

swarm = Swarm(agents=[researcher, analyst], entry_agent=researcher)
result = swarm("Analyse the market size of AI in 2030")
```

---

## Conversation Management

### Sliding window (default)

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

# param is 'window_size' (not 'max_messages'), default 40
agent = Agent(
    conversation_manager=SlidingWindowConversationManager(window_size=20),
)
```

### Summarising (condenses old messages)

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

# summary_ratio: fraction of oldest messages to summarise (default 0.3)
# preserve_recent_messages: minimum recent messages always kept (default 10)
agent = Agent(
    conversation_manager=SummarizingConversationManager(
        summary_ratio=0.4,
        preserve_recent_messages=10,
    ),
)
```

---

## Async Usage

Every synchronous method has an async counterpart:

```python
import asyncio
from elsai import Agent

async def main():
    agent = Agent()
    result = await agent.invoke_async("Hello!")
    print(result)

asyncio.run(main())
```

---

## Cancellation

```python
import threading, time
from elsai import Agent

agent = Agent()

def run():
    result = agent("Write a 10,000-word essay on history")
    print(result.stop_reason)  # "cancelled"

t = threading.Thread(target=run)
t.start()
time.sleep(2)
agent.cancel()
t.join()
```

---

## Snapshots

```python
# Save
snapshot = agent.take_snapshot(preset="session")
data = snapshot.to_dict()   # serialise to dict/JSON

# Restore
from elsai.types._snapshot import Snapshot
agent2 = Agent()
agent2.load_snapshot(Snapshot.from_dict(data))
```

---

## AgentResult

```python
result = agent("Hello!")
result.stop_reason          # "end_turn" | "max_tokens" | "cancelled" | ...
result.message              # the final assistant message dict
result.metrics.total_tokens # token usage
result.structured_output    # populated when structured_output_model was passed
str(result)                 # the text of the final message
```

---

## Common Patterns

### Multi-turn conversation

```python
agent = Agent()
agent("My name is Alice")
agent("I like Python")
response = agent("Summarise what you know about me")
```

### Inject messages directly

```python
from elsai.types.content import Message

agent = Agent(messages=[
    {"role": "user",      "content": [{"text": "Hello"}]},
    {"role": "assistant", "content": [{"text": "Hi! How can I help?"}]},
])
```

### Pass images

```python
import base64, pathlib

image_bytes = pathlib.Path("photo.jpg").read_bytes()
encoded = base64.b64encode(image_bytes).decode()

result = agent([{
    "image": {
        "format": "jpeg",
        "source": {"bytes": encoded},
    }
}, {
    "text": "Describe this image"
}])
```

---

## Key Rules When Writing Elsai Code

1. **`@tool` docstrings are mandatory** — the model reads them to decide when to call the tool.
2. **Return plain values from tools** — strings, dicts, Pydantic models all work; the SDK wraps them.
3. **Use `agent.cleanup()`** after any agent that uses MCP tools.
4. **`agent_id` drives session keys** — use stable, unique IDs per logical agent.
5. **`name` + `description` matter for agent-as-tool** — they become the tool's description the orchestrator sees.
6. **Hooks mutate events, not return values** — set `event.cancel_tool`, `event.retry`, `event.resume`, etc.
7. **Swarm handoffs are prompt-driven** — tell each agent when to hand off to others in the system prompt.
8. **Graph edges propagate output** — each node receives the combined output of its upstream nodes.
9. **`session_manager` + `agent_id` are both required** for sessions to persist correctly.
10. **All models are interchangeable** — swap the `model=` argument without changing anything else.