# More Useful Env Extensions (https://www.agentenvframework.com/docs/environments/useful-env-extensions)

> Some extensions we have found useful

## Forced Errors

An animation of `email` with forced errors: every third call to `send` fails.

1. The harness posts to `email`’s `set_errors`: from now on, every third call to `send` fails with a rate limit.
2. The agent’s first two `send` calls go through, and `email` counts them.
3. The third raises inside `email`, and the agent gets a tool error: `Error executing tool send: rate limit exceeded, retry later`.
4. The agent tries again and the fourth call goes through, so a task can check that the agent recovers instead of giving up.

The harness’s requests and what they answer:

```http
POST /svc/mcp-email/agentenv/ext/set_errors {"tool_name": "send", "every_nth": 3, "error_type": "rate_limit"}
→ 200 {"tool_name": "send", "every_nth": 3, "error_rate": 0.0, "error_type": "rate_limit"}
```

The agent’s calls:

- `send {"to": "sam.lee@example.com", "subject": "Q3 numbers", …}`: `{"sent": 1}`
- `send {"to": "dana@example.com", "subject": "Offsite", …}`: `{"sent": 2}`
- `send {"to": "priya@example.com", "subject": "Q3 review", …}`: tool error `Error executing tool send: rate limit exceeded, retry later`
- `send {"to": "priya@example.com", "subject": "Q3 review", …}`: `{"sent": 3}`

`urn:agentenv:set-errors/v1` makes a tool fail, to test how an agent copes with a flaky API. Here
`every_nth` fails every third call, and `error_rate` would fail a share of calls at random instead.
An exception in a tool reaches the agent as a tool error with its message.

```python title="email/server.py"
import random

from agentenv_protocol import AgentEnvEnvironment, environment_card, extension, tool

ERRORS = {"rate_limit": "rate limit exceeded, retry later", "timeout": "request timed out", "unavailable": "service unavailable"}


@environment_card(name="email")
class EmailEnv(AgentEnvEnvironment):
    def __init__(self) -> None:
        self.inbox: list[dict] = []
        self.sent: list[dict] = []
        self.faults: dict[str, dict] = {}

    @extension(uri="urn:agentenv:set-errors/v1", description="Make a tool fail.")
    async def set_errors(self, tool_name: str, every_nth: int = 0, error_rate: float = 0.0, error_type: str = "rate_limit") -> dict:
        self.faults[tool_name] = {"every_nth": every_nth, "error_rate": error_rate, "error_type": error_type, "calls": 0}
        return {"tool_name": tool_name, "every_nth": every_nth, "error_rate": error_rate, "error_type": error_type}

    def _fault(self, tool_name: str) -> None:
        fault = self.faults.get(tool_name)
        if fault is None:
            return
        fault["calls"] += 1
        nth = fault["every_nth"]
        failing = fault["calls"] % nth == 0 if nth else random.random() < fault["error_rate"]
        if failing:
            raise RuntimeError(ERRORS[fault["error_type"]])

    @tool()
    async def send(self, to: str, subject: str, body: str) -> dict:
        """Send an email."""
        self._fault("send")
        self.sent.append({"to": to, "subject": subject, "body": body})
        return {"sent": len(self.sent)}
```

## The Acting User

An animation of `email` acting as the user the harness sets.

1. `email` acts as `alex@example.com` until told otherwise, so its `email_search` sees only mail to Alex.
2. The harness posts `priya@example.com` to `set_acting_user`, and `email` now acts as Priya.
3. The agent’s `email_search` finds Priya’s mail, and its `send` goes out from Priya. The agent never chose who to be.
4. Set to `jordan@example.com`, the same calls search Jordan’s mail and send as Jordan.
5. A user `email` doesn’t know answers 500 `extension_failed`, and `email` keeps acting as Jordan, so a task can’t run as the wrong person.

The harness’s requests and what they answer:

```http
POST /svc/mcp-email/agentenv/ext/set_acting_user {"user_email": "priya@example.com"}
→ 200 {"user_email": "priya@example.com"}
POST /svc/mcp-email/agentenv/ext/set_acting_user {"user_email": "jordan@example.com"}
→ 200 {"user_email": "jordan@example.com"}
POST /svc/mcp-email/agentenv/ext/set_acting_user {"user_email": "nobody@example.com"}
→ 500 {"ok": false, "error": {"code": "extension_failed", "message": "unknown user: nobody@example.com"}}
```

The agent’s calls:

- `email_search {"query": "Q3"}`: `Q3 numbers, from dana@example.com`
- `send {"to": "dana@example.com", "subject": "Re: Q3 numbers", …}`: `{"sent": 1}`
- `email_search {"query": "Q3"}`: `Q3 plan, from sam.lee@example.com`
- `send {"to": "sam.lee@example.com", "subject": "Re: Q3 plan", …}`: `{"sent": 2}`

`urn:agentenv:set-acting-user/v1` sets who the server acts as, so one env can put the agent in
different people's shoes. Here it decides whose mail `email_search` finds and who `send` sends as.
Raising for a user the server doesn't know answers 500, so a task stops instead of running as the
wrong person.

```python title="email/server.py (the same EmailEnv)"
USERS = {"alex@example.com", "priya@example.com", "jordan@example.com"}


@environment_card(name="email")
class EmailEnv(AgentEnvEnvironment):
    def __init__(self) -> None:
        ...
        self.user_email = "alex@example.com"

    @extension(uri="urn:agentenv:set-acting-user/v1", description="Act as this user.")
    async def set_acting_user(self, user_email: str) -> dict:
        if user_email not in USERS:
            raise ValueError(f"unknown user: {user_email}")
        self.user_email = user_email
        return {"user_email": user_email}

    @tool()
    async def email_search(self, query: str) -> dict:
        """Find the acting user's emails whose subject or body contains the query."""
        q = query.lower()
        mine = [e for e in self.inbox if e["to"] == self.user_email]
        return {"emails": [e for e in mine if q in (e["subject"] + " " + e["body"]).lower()]}

    @tool()
    async def send(self, to: str, subject: str, body: str) -> dict:
        """Send an email as the acting user."""
        self._fault("send")
        self.sent.append({"from": self.user_email, "to": to, "subject": subject, "body": body})
        return {"sent": len(self.sent)}
```

## Async Waits

An animation of `reports` with an async wait on `reports_submit`.

1. The harness posts to `reports`’ `set_async_wait`: a job that `reports_submit` starts takes 30 to 90 seconds.
2. `reports_submit` answers at once with `job_1`, pending, and `reports` draws its wait: 45 seconds.
3. The agent polls `reports_status` at 15 and 30 seconds, and the job is still pending.
4. At 45 seconds the report is ready. The agent had to wait and poll for it, as it would for a real report.

The harness’s requests and what they answer:

```http
POST /svc/mcp-reports/agentenv/ext/set_async_wait {"tool_name": "reports_submit", "min_seconds": 30, "max_seconds": 90}
→ 200 {"tool_name": "reports_submit", "wait_range_seconds": [30, 90]}
```

The agent’s calls:

- `reports_submit {"query": "Q3 revenue"}` at 0 s: `{"job_id": "job_1", "status": "pending"}`
- `reports_status {"job_id": "job_1"}` at 15 s: `{"job_id": "job_1", "status": "pending"}`
- `reports_status {"job_id": "job_1"}` at 30 s: `{"job_id": "job_1", "status": "pending"}`
- `reports_status {"job_id": "job_1"}` at 45 s: `{"job_id": "job_1", "status": "ready", "rows": 12}`

`urn:agentenv:set-async-wait/v1` makes a tool's work take time, so an agent has to wait and poll as it
would against a real system. Here each job `reports_submit` starts is ready after a wait drawn between
`min_seconds` and `max_seconds`. A server that [follows the virtual
clock](https://www.agentenvframework.com/docs/environments/virtual-clock.md#follow-the-clock-in-a-server) can count that wait in virtual
seconds instead, so a fast clock shortens it.

```python title="reports/server.py"
import random
import time

from agentenv_protocol import AgentEnvEnvironment, environment_card, extension, tool


@environment_card(name="reports")
class ReportsEnv(AgentEnvEnvironment):
    def __init__(self) -> None:
        self.waits: dict[str, tuple[float, float]] = {}
        self.jobs: dict[str, dict] = {}

    @extension(uri="urn:agentenv:set-async-wait/v1", description="Make a tool's jobs take this long.")
    async def set_async_wait(self, tool_name: str, min_seconds: float, max_seconds: float | None = None) -> dict:
        self.waits[tool_name] = (min_seconds, max_seconds if max_seconds is not None else min_seconds)
        return {"tool_name": tool_name, "wait_range_seconds": list(self.waits[tool_name])}

    @tool()
    async def reports_submit(self, query: str) -> dict:
        """Start a report; poll reports_status for it."""
        wait = random.uniform(*self.waits.get("reports_submit", (0, 0)))
        job_id = f"job_{len(self.jobs) + 1}"
        self.jobs[job_id] = {"query": query, "ready_at": time.monotonic() + wait}
        return {"job_id": job_id, "status": "pending"}

    @tool()
    async def reports_status(self, job_id: str) -> dict:
        """Whether a report is ready, and its rows once it is."""
        job = self.jobs[job_id]
        if time.monotonic() < job["ready_at"]:
            return {"job_id": job_id, "status": "pending"}
        return {"job_id": job_id, "status": "ready", "rows": 12}
```

## In a Task

In a task, the `apply_server_config` step calls these extensions before the agent runs, as
[Env control](https://www.agentenvframework.com/docs/tasks/more-steps.md#env-control) describes.