# Creating your own agent (https://www.agentenvframework.com/docs/agents/creating)

> Write an agent as an AgentEnvAgent that answers A2A tasks with Claude and the tools of the envs it is given, calling itself on sub-tasks

1. Write the agent as an `AgentEnvAgent`: its identity, its config, its extensions and `run()`.
2. Deployed with `--sandbox local`, the agent serves the card the SDK built from the class, with its A2A endpoint and each extension’s.
3. The framework posts an env’s MCP URL to `/ext/mcp-config`. It reaches `run()` as `request.mcp_servers`.
4. A `message/send` to `/a2a` becomes one call to `run()`, which offers Claude the env’s tools and `recurse`.
5. Claude calls `recurse`, and a fresh call of itself, one level deeper, searches with `email_search`. With its answer, Claude picks `send`. `run()` calls each env tool over MCP.
6. `run()` returns a `TaskResult`: the reply, the usage, and a trajectory the framework reads back.
7. Deploy the same agent with `--sandbox modal_vm`, then on a provider of your own: a new instance each time, with its own id and A2A URL, and the same card.

Parts of the scene:

- **Your agent**: A subclass of `AgentEnvAgent` from `agentenv-framework-protocol[agent]`. `@a2a_agent` says what its card lists, and `run()` does the work of each task.
- **Identity**: `AgentIdentity` gives the agent’s name, description and version. They become the card’s `name`, `description` and `version`.
- **Configuration**: `ClaudiusRLMConfig` subclasses `AgentConfig`. Its fields become the agent-config extension, and the framework sets the ones a task names, such as `model`, `effort` and `max_turns`, before each prompt.
- **Extensions**: `MCP_CONFIG_V1` takes the environments to use, and `TRAJECTORY_V1` hands back what the agent did. The SDK implements both, so declaring them is enough.
- **run()**: Called once for each A2A task. It gets a `TaskRequest`, calls Claude with the env’s tools and `recurse` until Claude answers, and returns a `TaskResult`, or `TaskResult.failure(code, message)` when it cannot finish.
- **The agent card**: Served at `/.well-known/agent-card.json`. It gives the agent’s name, its A2A endpoint and every extension it supports, with their endpoints and the fields they take. Every instance serves the same card.
- **The A2A endpoint**: `/a2a` takes JSON-RPC. A `message/send` starts a task and answers with it once `run()` returns: its state, the reply and a data part with the usage.
- **The deploy**: `agent-env a2a-agent deploy --id claudius` starts an instance of a registered agent. `--sandbox` picks where; without it, `[sandbox] agent_default`, else `local`.
- **An agent instance**: One deployment of the agent, with its own id, the agent id plus eight random characters, and its own A2A URL. Deploying again starts another; the agent itself never changes.
- **local**: Runs the agent as a container on your machine’s Docker, with its A2A port published on `localhost`. It does not enforce the TTL.
- **modal_vm**: Runs the agent in a Modal VM, which loads the agent’s image and starts it there. Modal pulls the image from your image store, so it needs a store Modal can reach, and your Modal credentials.
- **Your sandbox provider**: Subclass `SandboxProvider` and register it under `[sandbox.providers.<name>]` in `config.toml`, or through the `agent_env.sandbox_providers` entry point. Then deploy with `--sandbox <name>`.
- **mcp-config**: `POST /ext/mcp-config` with an MCP `url` and a `name` registers an environment. A `deploy_agent` step posts one for each env it is given.
- **trajectory**: `POST /ext/trajectory` with a task’s `id` returns the native trajectory that `run()` returned for it. A `prompt_agent` step reads it once the agent replies.
- **The TaskRequest**: What one task gives `run()`: the message in `parts`, the typed `config`, the environments in `mcp_servers` and any installed `skills`.
- **The TaskResult**: The reply, a `Usage` with the number of tool calls, and a native trajectory in a format the agent names. A success needs at least one text, file or data part.
- **The framework**: Or anything else that speaks A2A. It reads the card, hands the agent environments through mcp-config, sends tasks to `/a2a` and reads trajectories back.
- **Claude**: Claude Opus 5.5 through the Anthropic SDK, at `LITELLM_BASE_URL` with `LITELLM_API_KEY`, which the framework sets when it deploys the agent. A `recurse` call starts a fresh call of Claude one level deeper, which returns only its answer.
- **The environment**: An instance of `email`, reached over MCP at the URL mcp-config registered. The agent lists its tools, `email_search` and `send`, and calls them.

You write an agent as a subclass of `AgentEnvAgent`, from `agentenv-framework-protocol[agent]`. This
one, the agent these pages follow, is Claude as a recursive language model: it answers each task
with the tools of the environments the framework hands it, and hands sub-tasks to fresh calls of
itself:

```python title="claudius/agent.py"
import os
from contextlib import AsyncExitStack

from anthropic import AsyncAnthropic
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

from agentenv_protocol.a2a_agent import (
    MCP_CONFIG_V1,
    TRAJECTORY_V1,
    AgentConfig,
    AgentEnvAgent,
    AgentIdentity,
    TaskRequest,
    TaskResult,
    TextPart,
    Usage,
    a2a_agent,
)

RECURSE = {
    "name": "recurse",
    "description": (
        "Hand a self-contained sub-task to a fresh call of yourself, with its own context and the same "
        "tools. It returns only its answer, so use it for work whose details you do not need to keep."
    ),
    "input_schema": {
        "type": "object",
        "properties": {"task": {"type": "string", "description": "The sub-task, with everything it needs."}},
        "required": ["task"],
    },
}


class ClaudiusRLMConfig(AgentConfig):
    model: str = "claude-opus-5-5"
    system_prompt: str = "You work an email inbox. Use the tools to do what you are asked."
    effort: str = "high"
    max_turns: int = 10
    max_depth: int = 2


class Stopped(Exception):
    """A call that ended without an answer."""


@a2a_agent(
    identity=AgentIdentity(
        name="claudius",
        description="Claude that works an inbox through the tools of the envs it is given, and calls itself on sub-tasks.",
        version="1.0.0",
    ),
    config=ClaudiusRLMConfig,
    extensions=(MCP_CONFIG_V1, TRAJECTORY_V1),
)
class ClaudiusRLMAgent(AgentEnvAgent):
    async def run(self, request: TaskRequest[ClaudiusRLMConfig]) -> TaskResult:
        config = request.config
        claude = AsyncAnthropic(
            base_url=os.environ["LITELLM_BASE_URL"].removesuffix("/v1"), api_key=os.environ["LITELLM_API_KEY"]
        )
        calls = []

        async def solve(task: str, depth: int) -> str:
            tools = env_tools + ([RECURSE] if depth < config.max_depth else [])
            messages = [{"role": "user", "content": task}]
            for _ in range(config.max_turns):
                response = await claude.messages.create(
                    model=config.model,
                    max_tokens=16000,
                    system=config.system_prompt,
                    tools=tools,
                    messages=messages,
                    output_config={"effort": config.effort},
                )
                if response.stop_reason == "refusal":
                    raise Stopped("refusal", "Claude declined the task")
                if response.stop_reason != "tool_use":
                    return "".join(block.text for block in response.content if block.type == "text")
                messages.append({"role": "assistant", "content": response.content})
                results = []
                for block in response.content:
                    if block.type != "tool_use":
                        continue
                    if block.name == RECURSE["name"]:
                        output = await solve(block.input["task"], depth + 1)
                    else:
                        result = await sessions[block.name].call_tool(block.name, block.input)
                        output = "\n".join(c.text for c in result.content if c.type == "text")
                    calls.append({"depth": depth, "tool": block.name, "input": block.input, "output": output})
                    results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})
                messages.append({"role": "user", "content": results})
            raise Stopped("max_turns", f"Still calling tools after {config.max_turns} turns at depth {depth}")

        async with AsyncExitStack() as stack:
            # One MCP session per env the framework handed over, and every tool they offer.
            sessions, env_tools = {}, []
            for server in request.mcp_servers.values():
                read, write, _ = await stack.enter_async_context(
                    streamablehttp_client(server["url"], headers=server.get("headers"))
                )
                session = await stack.enter_async_context(ClientSession(read, write))
                await session.initialize()
                for tool in (await session.list_tools()).tools:
                    sessions[tool.name] = session
                    env_tools.append(
                        {"name": tool.name, "description": tool.description or "", "input_schema": tool.inputSchema}
                    )

            prompt = "\n".join(part.text for part in request.parts if isinstance(part, TextPart))
            try:
                answer = await solve(prompt, depth=0)
            except Stopped as stop:
                return TaskResult.failure(*stop.args)

        return (
            TaskResult.builder()
            .succeeded()
            .add_text(answer)
            .usage(Usage(tool_call_count=len(calls)))
            .native_trajectory(format="claudius/v1", payload=calls)
            .build()
        )


if __name__ == "__main__":
    ClaudiusRLMAgent().serve()
```

## The run method

Each A2A task the agent receives is one call to `run()`. The request carries the message in `parts`,
the agent's `config`, the environments to use in `mcp_servers` and any installed `skills`.
`ClaudiusRLMAgent` connects to each environment over MCP and hands the prompt to `solve()`, which
calls Claude with the environments' tools until Claude answers. `run()` returns a `TaskResult` with
the answer, the usage and a trajectory of every tool call. When it cannot finish, it returns
`TaskResult.failure(code, message)` instead.

Claude is called through the Anthropic SDK, at `LITELLM_BASE_URL` with `LITELLM_API_KEY`, which the
framework sets when it deploys the agent, as [Deploying your
agent](https://www.agentenvframework.com/docs/agents/deploying.md#point-it-at-a-model) shows. The SDK adds `/v1` to a base URL itself, so
`run()` strips it.

## Configuration

`ClaudiusRLMConfig` lists the settings the agent takes, with their defaults. Before each prompt, the
framework sends the ones a task sets, such as `model`, `effort` or `max_turns`, and `run()` reads
them as `request.config`. `effort` goes to Claude in `output_config`: Claude Opus 5.5 always thinks,
and effort sets how much.

## Identity and extensions

`@a2a_agent` declares what the agent card says: the agent's identity, its configuration and the
extensions it supports. Extensions are how the framework works with an agent beyond sending it
messages. `ClaudiusRLMAgent` declares two that the SDK implements for it: `MCP_CONFIG_V1`, which
hands it environments, and `TRAJECTORY_V1`, which reads back what it did. [Useful A2A
extensions](https://www.agentenvframework.com/docs/agents/extensions.md) covers the rest.

## Wrapping a harness

A harness you already use, such as Claude Code, becomes an agent the same way: its `run()` writes
`request.mcp_servers` into the harness's MCP configuration, starts the harness with the prompt and
returns its final message. The framework does not ship these wrappers.

To put `ClaudiusRLMAgent` in a task, you deploy it. [Deploying your agent](https://www.agentenvframework.com/docs/agents/deploying.md)
packages it, runs it in a sandbox and hands it an environment.