# elsai Agent SDK — Cursor Rules

You are an expert in the **elsai Agent SDK** for building AI agents in Python.

## Installation

```bash
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ elsai-agents==0.3.0
pip install --extra-index-url https://core-packages.elsai.ai/root/elsai-model/ elsai-model==2.0.0
# A2A Protocol v1.0 (optional)
pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[a2a]==0.3.0"
```

## Core Imports

```python
from elsai import Agent, tool
from elsai_model.anthropic import AnthropicModel
from elsai_model.bedrock import BedrockModel
from elsai_model.gemini import GeminiModel
from elsai_model.litellm import LiteLLMModel
from elsai_model.ollama import OllamaModel
from elsai_model.openai import OpenAIModel
from elsai.hooks import (
    BeforeModelCallEvent, AfterModelCallEvent,
    BeforeToolCallEvent, AfterToolCallEvent,
    BeforeInvocationEvent, AfterInvocationEvent,
    AgentErrorEvent, MessageAddedEvent,
)
from elsai.multiagent import GraphBuilder, Swarm
from elsai.execution import LocalSandboxBackend, Sandbox, workspace_id_for_run
from elsai.plugins.sandbox import SandboxPlugin
from elsai.session import FileSessionManager, S3SessionManager
from elsai.types import ToolContext
```

## Minimal Agent

```python
from elsai import Agent

agent = Agent()  # defaults to Amazon Bedrock
result = agent("Explain quantum computing")
print(result)

# With a specific model
agent = Agent(model="us.amazon.nova-pro-v1:0")
agent = Agent(model=AnthropicModel(model_id="claude-sonnet-4-6"))
agent = Agent(model=OpenAIModel(model_id="gpt-4o"))
```

## Tool Creation Rules

- **ALWAYS write a docstring** — the model reads it to decide when/how to call the tool
- Type-annotate ALL parameters
- Return `str`, `dict`, `int`, `float`, Pydantic model, or any JSON-serialisable value
- For error: `return {"status": "error", "content": [{"text": "error message"}]}`

```python
from elsai import tool

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result.

    Args:
        expression: A valid Python math expression, e.g. '2 + 2 * 3'.
    """
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

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

@tool
async def fetch_url(url: str) -> str:
    """Fetch the content of a URL."""
    ...

# With ToolContext (access agent state and invocation state)
@tool(context=True)
def context_aware(question: str, tool_context: ToolContext) -> str:
    """Tool with access to agent context."""
    agent = tool_context["agent"]
    return f"Agent {agent.name} asked: {question}"
```

## Agent with Tools

```python
agent = Agent(
    model="us.amazon.nova-pro-v1:0",
    system_prompt="You are a helpful assistant with access to tools.",
    tools=[calculate, search, fetch_url],
)

# Direct tool call (bypass LLM)
result = agent.tool.calculate(expression="42 * 7")

# Natural language
result = agent("What is 42 times 7?")
```

## Hooks

```python
agent = Agent()

# Type-hint inference — event type is inferred from parameter annotation
def log_call(event: BeforeModelCallEvent) -> None:
    print(f"Calling model for: {event.agent.name}")

def block_tool(event: BeforeToolCallEvent) -> None:
    if event.tool_use["name"] == "sensitive_tool":
        event.cancel_tool = "Blocked by policy"

def auto_retry(event: AfterToolCallEvent) -> None:
    if event.exception:
        event.retry = True

agent.add_hook(log_call)
agent.add_hook(block_tool)
agent.add_hook(auto_retry)

# Or pass at construction
from elsai.agent import AgentConfig

agent = Agent(config=AgentConfig(hooks=[log_call, block_tool]))
```

## Multi-Agent: Graph (deterministic)

```python
from elsai.multiagent import GraphBuilder

planner  = Agent(system_prompt="Plan the task step by step.", config=AgentConfig(name="planner"))
executor = Agent(system_prompt="Execute each step.", config=AgentConfig(name="executor"))
reviewer = Agent(system_prompt="Review and summarise.", config=AgentConfig(name="reviewer"))

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")
print(result.results["review"].message)
```

## Multi-Agent: Swarm (autonomous handoff)

```python
from elsai.multiagent import Swarm

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

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

## Agent-as-Tool

```python
sub_agent = Agent(
    tools=[web_search],
    config=AgentConfig(
        name="researcher",
        description="Searches and retrieves information on any topic",
    ),
)

orchestrator = Agent(tools=[sub_agent.as_tool()])
result = orchestrator("Research the history of Python and write a summary")
```

**Agent-as-tool concurrency:** `tools=[sub_agent]` auto-wraps with `mode="reject"`. Parallel calls to the **same** specialist → `sub_agent.as_tool(mode="spawn")`. OK to wait on a shared agent → `mode="queue"`.

## Sessions (Persistence)

```python
from elsai.session import FileSessionManager, S3SessionManager

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

agent("My name is Alice")
# Restart process — conversation restored automatically
agent("What is my name?")  # → "Alice"

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

## Structured Output

```python
from pydantic import BaseModel

class Summary(BaseModel):
    title: str
    key_points: list[str]
    conclusion: str

agent = Agent()
result = agent(
    "Summarise the history of Python",
    structured_output_model=Summary,
)
summary: Summary = result.structured_output
```

## Async

```python
import asyncio

async def main():
    agent = Agent()
    result = await agent.invoke_async("Hello!")
    async for event in agent.stream_async("Tell me a story"):
        if "data" in event:
            print(event["data"], end="")

asyncio.run(main())
```

## Sandbox Execution

Isolated workspace runs — the LLM stays on the host; shell and file operations use sandbox plugin tools.

```python
import asyncio
from pathlib import Path
from uuid import uuid4
from elsai import Agent
from elsai.agent import AgentConfig
from elsai.execution import LocalSandboxBackend, Sandbox, workspace_id_for_run
from elsai.plugins.sandbox import SandboxPlugin

async def main() -> None:
    async with Sandbox(
        workspace_id=workspace_id_for_run(str(uuid4())),
        backend=LocalSandboxBackend(Path("/tmp/elsai-sandbox")),
        ttl_seconds=3600,
        seed_from=Path("./my-repo"),
    ) as sandbox:
        agent = Agent(config=AgentConfig(sandbox=sandbox, plugins=[SandboxPlugin()]))
        result = await agent.invoke_async("List project files and summarize README")
        print(result.message)

asyncio.run(main())
```

- **`SandboxPlugin`** registers `execute`, `read_file`, `write_file`, `list_dir` (auto-added on attach when `sandbox=` is set)
- **Multi-agent:** attach planner, coder, and reviewer to the same `Sandbox` instance
- **Production backends:** `AgentCoreSandboxTemplate`, `DaytonaSandboxTemplate` from `elsai_tools.sandbox` (requires `elsai-agents-tools==0.2.0`)
- **`ttl_seconds`** is required on `Sandbox(...)`; prefer `async with Sandbox(...) as sandbox:`

## Key Rules

1. ALWAYS write docstrings for `@tool` functions — the LLM reads them.
2. Call `agent.cleanup()` when done with MCP tools.
3. Both `agent_id` AND `session_manager` are required for session persistence.
4. `name` + `description` on Agent matter when using `agent.as_tool()`.
5. **Agent-as-tool concurrency:** `tools=[agent]` → `mode="reject"`. Same specialist in parallel → `agent.as_tool(mode="spawn")`. OK to wait → `mode="queue"`.
6. In Swarm: handoff logic lives in the system prompt ("hand off to X when…").
7. In Graph: each node receives the combined output of all upstream nodes.
8. Hooks mutate event attributes (`event.cancel_tool`, `event.retry`, `event.resume`).
9. All model providers share the same Agent API — just swap `model=`.
10. In **sandbox mode**, use sandbox plugin tools — host `shell`, `file_read`, `file_write`, `python_repl`, and `environment` are blocked.