# AG2 Development Guidelines

## AI-assisted contribution policy

Before opening a PR, read and follow `.github/AI_POLICY.md`.

- Do not open PRs with unverified AI-generated code or text.
- Ensure the PR description explains the real problem or use case and accurately reflects the diff.
- Include validation and testing information in the PR body.
- Be prepared to explain and revise the contribution in response to reviewer questions.
- Write the PR description using `.github/PULL_REQUEST_TEMPLATE.md`. Keep its section headings (`## Why are these changes needed?`, `## Related issue number`, `## Checks`, `## AI assistance`), fill each one in, and only check a checklist box once it is actually true.

## Architecture Decision Records (ADR)

Cross-cutting and hard-to-reverse design decisions are recorded in `docs/adr/`, sequentially numbered (`0001-*.md`, `0002-*.md`, …) with `status` / `date` frontmatter and a short Context / Decision / Consequences body.

- **Consult them before changing established public API or architecture.** They explain *why* something is the way it is — e.g. `0003-eval-run-api-takes-agent-instances.md` records that the eval `run_*` API takes prebuilt `Agent` instances (not factories) and explicitly-built `Suite`s. If a change contradicts an ADR, supersede it rather than silently reverting the code.
- **Add one when a decision qualifies**: it is hard to reverse, surprising without context (a reader would assume the opposite), and the result of a real trade-off. Scan `docs/adr/` for the highest number and increment. Keep it short — recording *that* a decision was made and *why* is the value.

## Code Style Guidelines

- Do not use `from __future__ import annotations`.
- With `@contextmanager` / `@asynccontextmanager`, annotate the return type as `Generator[T]` / `AsyncGenerator[T]`, never `Iterator[T]` / `AsyncIterator[T]`. The decorator needs a real generator — it calls `throw()` / `athrow()` on it — so the iterator form is an under-specification that typeshed marks deprecated. Import them from `collections.abc` (not `typing`) and omit the default send type: `AsyncGenerator[None]`, not `AsyncGenerator[None, None]`.
- Do not use global variables or top-level side-effect function calls unless the user explicitly allows it.
- For filesystem paths, use `pathlib.Path` internally. Public signatures should accept `str | os.PathLike[str]`.
- Top-level imports from `ag2.*` are for common APIs that are broadly reusable across scenarios and core agent flows.
  Good: `ag2.[Input]` — common structures usable in `await agent.ask(Input())` and as tool results.
  Bad: `ag2.middleware.BaseMiddleware` — this is advanced/specialized and should be imported only when implementing custom middleware.
- Do not use function-level imports unless the user explicitly allows it.
  ```python
  # === BAD - import inside function ===
  def execute_tool():
      from .tool import Tool

      ...


  # === GOOD - top-level import ===
  from .tool import Tool


  def execute_tool(): ...
  ```
- Do not create nested functions inside runtime execution paths.
  ```python
  # === BAD - function will be created each call ===
  def execute_tool():
      def _inner_function():
          pass

      _inner_function()


  # === GOOD - function created once, executed each call ===
  def execute_tool():
      _inner_function()


  def _inner_function():
      pass


  # === GOOD - decorator executed import time, so we can use closure functions here ===
  def decorator(func):
      def wrapper():
          return func()

      return wrapper
  ```
- Do not perform side effects in initialization methods. Apply side effects only at runtime.
  ```python
  # === BAD - create directory in initial method ===
  class KnowledgeStore:
      def __init__(self, path: str | os.PathLike[str]) -> None:
          self.path = Path(path)
          # side effect - directory creation
          self.path.parent.mkdir(parents=True, exist_ok=True)

      def run(self) -> None: ...


  # === GOOD - create directory in runtime method ===
  class KnowledgeStore:
      def __init__(self, path: str | os.PathLike[str]) -> None:
          self.path = Path(path)

      def run(self) -> None:
          self.path.parent.mkdir(parents=True, exist_ok=True)
          ...
  ```

## Package Structure

`ag2/` is a protocol-driven async agent framework. Key modules:

| Module | Purpose | Key Exports |
|--------|---------|-------------|
| `agent.py` | Core agent loop and reply handling | `Agent`, `AgentReply` |
| `annotations.py` | Type annotations for dependency injection | `Context`, `Inject`, `Variable` |
| `context.py` | Runtime context (stream, dependencies, variables, prompt) | `Context` dataclass, `Stream` protocol |
| `stream.py` | In-memory event pub/sub | `MemoryStream`, `SubStream` |
| `events/` | Event types for the agent loop | `BaseEvent`, `ModelRequest`, `ModelResponse`, `ToolCallEvent`, `ToolResultEvent`, `Usage`, … |
| `config/` | LLM provider clients (see [below](#llm-provider-clients)) | `ModelConfig`, `LLMClient`, `AnthropicConfig`, `OpenAIConfig`, `GeminiConfig`, … |
| `tools/` | Ready-to-use tools | `tool`, `Toolkit`, `ToolResult`, `CodeExecutionTool`, `ShellTool`, `WebSearchTool`, … |
| `tools/types.py` | Tool abstractions & schema types | `Tool`, `FunctionTool`, `ClientTool`, `Toolkit`, `ToolSchema`, … |
| `tools/subagents/` | Agent-to-agent delegation | `subagent_tool`, `persistent_stream`, `StreamFactory`, `StreamHandle`, `DeliveryError` |
| `eval/` | Offline evaluation framework | `run`, `scorer`, `EvalTarget`, `Suite`, `Task`, `Trace`, `RunResult`, `Feedback`, `BudgetThresholds`, plus prebuilts under `eval.scorers` |
| `middleware/` | Request/response interception | `BaseMiddleware`, `Middleware`, `LoggingMiddleware`, `RetryMiddleware`, `TokenLimiter`, `HistoryLimiter`, `MiddlewareDescription`, … |
| `response/` | Structured output validation | `ResponseSchema`, `PromptedSchema`, `ResponseProto`, `response_schema` |
| `history.py` | Conversation history storage | `History`, `Storage`, `MemoryStorage` |
| `scopes.py` | Variable scopes across a delegation tree | `ScopedValue`, `Scope`, `ScopeError` |
| `hitl.py` | Human-in-the-loop hooks | — |
| `streams/` | Persistent stream backends (e.g. Redis) | — |

### Public API (`ag2`)

Top-level modules:
- `ag2` - top-level module with most basic functionality
- `ag2.types` - Type aliases and constants
- `ag2.config` - LLM provider clients (see [below](#llm-provider-clients))
- `ag2.tools` - Tool system — builtin + user-defined (see [below](#builtin-tools))
- `ag2.tools.subagents` - Agent-to-agent delegation (see [below](#subagent-delegation))
- `ag2.testing` - Testing utilities
- `ag2.middleware` - Request/response interception (see [below](#middleware))
- `ag2.observers` - Reusable observer implementations
- `ag2.eval` - Offline evaluation framework (datasets, scorers, runner, persistence)

Advanced modules:
- `ag2.events` - Event types for the agent loop
- `ag2.streams` - Persistent stream backends (e.g. Redis)
- `ag2.watch` - Watch system for triggering observers
- `ag2.knowledge` - Knowledge management
- `ag2.plugin` - Plugin system

### Re-export rules

All implementations must be re-exported from their public module's `__init__.py` and listed in `__all__`. If an implementation requires third-party dependencies, wrap the import in a `try/except ImportError` block and register a missing-dependency fallback so users get a clear install hint instead of an unexplained `ImportError` (see `ag2/config/__init__.py`, `ag2/middleware/builtin/__init__.py` as the reference pattern). Two fallbacks exist:

- **Core modules** use **optional dependencies** shipped as pyproject extras — fall back via `missing_optional_dependency`, which hints `pip install "ag2[<extra>]"`.
- **Extensions** (`ag2/extensions/`) are **not** shipped as extras — declare their third-party packages as **additional dependencies** and fall back via `missing_additional_dependency`, which hints the upstream package directly (e.g. `pip install "daytona>=0.171.0,<1"`).

Either way, add the third-party package to the `optionals` **dependency group** in `pyproject.toml` (PEP 735 — CI-only, never part of `pip install ag2`) so the test workflow installs it. Tests guarded by `pytest.importorskip("<package>")` are silently skipped otherwise, and CI still passes.

### Design principles

- **Protocols over inheritance**: `LLMClient`, `ModelConfig`, `Stream`, `Storage` are all `Protocol` classes — implementations satisfy them structurally. `Tool` is the exception: it is an `ABC` that tool kinds subclass (see [ADR 0002](docs/adr/0002-tool-composite-hierarchy.md)).
- **Async throughout**: all major operations (`ask`, tool execution, LLM calls) are async. Sync tool functions run via `sync_to_thread`.
- **Event-driven**: all agent-loop communication flows through the `Stream` as typed events.
- **Dependency injection**: all user-provided functions (tools, prompt hooks, HITL, etc.) use `Context`, `Inject`, and `Variable` annotations; resolution is handled by `fast_depends`.

## Builtin Tools

Builtin tools live in `ag2/tools/builtin/`. Each tool has:
- A `ToolSchema` dataclass (provider-neutral capability flag)
- A `Tool` class (constructs the schema, resolves Variables)

### API Design

- Use `version` as the public parameter name on Tool constructors for provider-versioned tools (e.g., `WebSearchTool(version="web_search_20260209")`). The schema field may use a more specific name internally (e.g., `web_search_version`) — the Tool maps between them.
- Tool constructor parameters that accept runtime values must also accept `Variable` for deferred context resolution (e.g., `max_uses: int | Variable | None`).
- Tools with no configurable parameters (e.g., `MemoryTool`, `CodeExecutionTool`) should still accept a `version` keyword argument to allow version pinning.
- Provider mappers in `ag2/config/{provider}/mappers.py` convert `ToolSchema` instances to provider-specific API dicts. Use `t.version` instead of hardcoding version strings.

### Adding a New Builtin Tool

1. Create `ag2/tools/builtin/{tool_name}.py` with a `ToolSchema` dataclass and `Tool` class.
2. Add mapper handling in every provider's mapper:
   - Supported: add an `elif isinstance(t, YourToolSchema)` branch returning the provider-specific dict.
   - Unsupported: the existing fallback `raise UnsupportedToolError(t.type, "provider")` handles it.
3. Add tests for every provider (see test guidelines below).
4. If the tool accepts `Variable` parameters, add 2 tests to `test/tools/test_resolve.py`: one resolving from context, one raising `KeyError` on missing.

## Subagent Delegation

Subagent tools live in `ag2/tools/subagents/` and are imported from `ag2.tools.subagents` (not re-exported from `ag2.tools`).

Application code delegates through `Agent.as_tool()` / `subagent_tool()` and
addresses an observed event's author through `event.origin.send()`.
This is the single public addressed-delivery entry point; each handle belongs
to one invocation. `run_task`, `TaskResult`, and `DelegationRegistry` remain internal;
framework code and tests import them from their implementation modules, not the
package's public exports.

Every descendant event reaches every ancestor. Inert events (`InertEvent`:
`UsageEvent` and every `TaskEvent`) arrive as themselves, so existing
type-based observer and stream subscriptions receive descendant costs and task
lifecycle directly. The first forwarding hop stamps their `path` (and
`task_id` / `agent_name` on `UsageEvent`); later hops keep it, and `UsageEvent`
keeps its source `event_id`. All other events arrive as inert `ChildEvent`
observations, which never trigger the ancestor's tools, model loop or HITL.
Accounting uses no generated rollups. See `docs/design/subagent-eventbus-v2.md`
and ADRs 0030–0032.

| File | Purpose |
|------|---------|
| `run_task.py` | Internal `run_task()`, `TaskResult` — execute an agent as a sub-task |
| `subagent_tool.py` | `subagent_tool()`, `StreamFactory` — wrap an agent as a callable tool |
| `persistent_stream.py` | `persistent_stream()` — `StreamFactory` that reuses a stream across calls |

### Agent.as_tool()

`Agent.as_tool(description, name?, stream?, middleware?)` is a convenience method that delegates to `subagent_tool()`. It creates a tool named `task_{agent.name}` with parameters `objective` (required) and `context` (optional).

Delegation is explicit: build each worker `Agent` with its own `config`, `prompt` and `tools`, then pass `worker.as_tool(...)` to the parent's `tools`. Nothing is inherited from the parent.

- **Recursion** is bounded only by what you hand the worker: do not give a worker delegate tools unless recursion is intended.
- **Concurrency**: the LLM fans out by issuing several delegate calls in one response; they run concurrently. To cap them, pass the same tool middleware to every delegate. It must be a function closing over the `asyncio.Semaphore`, not a middleware object: tool middleware is deep-copied on registration, so an object gets one copy per delegate and the cap silently splits.
- **Ad-hoc workers** composed by the LLM per objective: `dynamic_agent()` (`ag2.tools.dynamic`).

Each delegation runs on its own stream. Operational events reach ancestors as
`ChildEvent` observations; usage and lifecycle events reach every ancestor as
themselves, including events from grandchildren.

### Deprecated: `TaskConfig` / auto-injected `run_subtask` (removed in 1.4.0)

Do not use `TaskConfig`, `Agent(tasks=...)` or `run_subtask` / `run_subtasks` in new code, docs or examples. The legacy path still ships in `ag2/agent.py` until removal.

- Tests that exercise it on purpose carry `@pytest.mark.filterwarnings("ignore:TaskConfig, Agent:DeprecationWarning")` (or a module-level `pytestmark`); new tests must not depend on it.
- Removal in 1.4.0 covers: `TaskConfig` and its re-export from `ag2`, the `tasks=` parameter and `Agent.tasks`, `_build_subtask_toolkit` / `_spawn_subtask` / `_run_subtask` / `_task_slots_for`, the `agent.tasks` check in `ag2/ag_ui/capabilities.py`, those tests, and the deprecation notes in `website/docs/`.

### Deprecated: `Agent(assembly=...)` (removed in 1.4.0)

Register `Middleware(AssemblerMiddleware, policies=[...])` (plus `Middleware(HaltCheckMiddleware)` with `AlertPolicy`) in `middleware=` instead. Plugin, `add_policy()` and invocation-plugin policies join that assembler each turn.

- Tests that exercise it on purpose carry `@pytest.mark.filterwarnings("ignore:Agent\\(assembly=:DeprecationWarning")` (or a module-level `pytestmark`); new tests must not depend on it.
- Removal in 1.4.0 covers: the `assembly=` parameter and `ASSEMBLY_DEPRECATION`, the warning-filtered tests, and the deprecation notes in `website/docs/`. Keep `_attach_policies`: plugin policies still need it.

### Deprecated: `Agent(knowledge=...)` (removed in 1.4.0)

Use `plugins=[KnowledgePlugin(store, ...)]` (`ag2.knowledge`) instead. At most one `KnowledgePlugin` applies to an invocation; combining it with `knowledge=` is rejected. See ADR 0033.

- Tests that exercise it on purpose carry `@pytest.mark.filterwarnings("ignore:Agent\\(knowledge=:DeprecationWarning")` (or a module-level `pytestmark`); new tests must not depend on it.
- Removal in 1.4.0 covers: the `knowledge=` parameter and `KNOWLEDGE_DEPRECATION`, the `LegacyKnowledgeAdapter` block in `Agent.__init__`, the `knowledge=` branch of `validate_knowledge_middleware`, `ag2/knowledge/_legacy.py`, the bound-store (`legacy_store`) branches in knowledge tools and middleware factories, the warning-filtered tests, and the deprecation notes in `website/docs/`. Keep `KnowledgeConfig` and its exports, `ScopedMiddlewareFactory`, and `_additional_tools` (legacy task tools still use it until `TaskConfig` is removed).

### Deprecated: `Agent.task()` (removed in 1.4.0)

Construct `Task(owner_id=agent.name, spec=TaskSpec(title=..., description=..., payload=..., capability=...), ...)` from `ag2.task` instead; pass `context`, `ttl_seconds`, `checkpoint_store` and `resume_from` directly to `Task`. For model runs, pass `capability=` / `ttl_seconds=` to `ask` / `run` / `resume`. See ADRs 0034 and 0035.

- Tests that exercise it on purpose assert the warning with `pytest.warns(DeprecationWarning)` (see `test/agent/test_task_deprecation.py`); new tests must not depend on it.
- Removal in 1.4.0 covers: `Agent.task` and `AGENT_TASK_DEPRECATION`, the `deprecated-factory` parametrisations and `test_deprecated_factory_still_fails_the_task_on_error` in `test/agent/test_task_deprecation.py`, and the deprecation notes in `website/docs/`. Keep `Task`, `TaskSpec` and `TaskInject`.

### persistent_stream

`persistent_stream()` returns a `StreamFactory` that gives the same agent a consistent stream across multiple invocations within a context. It stores the stream ID in `context.dependencies` keyed by `ag:{agent.name}:stream`, and reuses the parent stream's storage backend.

Use it when sub-task history should accumulate across calls rather than starting fresh each time:

```python
agent.as_tool(description="...", stream=persistent_stream())
```

### Context flow in run_task

| What | Behavior | Why |
|------|----------|-----|
| Dependencies | Shallow-copied from the parent. Request values (`ask(dependencies=...)`, set at runtime) override the worker's own; the parent's configuration (constructor, plugin defaults) only fills keys the worker lacks | Isolated at the top level; mutable values are still shared by reference. Configuration vs request is tracked by identity under `CONFIGURED_DEPENDENCIES_CONTEXT_KEY`, published per turn by `_turn_scope`, so a worker's `KnowledgePlugin` store is not replaced by the parent's while a per-request store still reaches it. Treat dependencies as read-only inside subtasks. |
| Variables | Serializable values: plain/copied values are deep-copied, local values are excluded, and root-declared shared bindings remain shared | Copied values stay branch-local; shared mutations require application-supplied locks in `dependencies`. `ask` copies the caller's mapping, so read results from `reply.context.variables`. |
| History | Fresh stream per call | Clean context; LLM passes relevant info via `context` parameter |
| Tools | The worker's own `tools=`; nothing is inherited from the parent | The worker's capabilities are explicit and reviewable at construction |

## LLM Provider Clients

Provider clients live in `ag2/config/{provider}/`. Each provider has at least three files:
- `config.py` — a `@dataclass(slots=True)` implementing the `ModelConfig` protocol
- `{provider}_client.py` — a concrete class satisfying the `LLMClient` protocol (async `__call__`)
- `mappers.py` — pure functions for converting messages, tools, response schemas, and usage between internal and provider-specific formats

### Client conventions

- The constructor takes connection params (api_key, base_url, timeout, …) plus a `CreateOptions` TypedDict for generation params. It wraps the provider's async SDK client.
- `__call__` converts messages/tools via mappers, calls the provider API, normalises the response into `ModelResponse` with `Usage`.
- Streaming: emit `ModelMessageChunk` / `ModelReasoning` events via `context.send()` while accumulating the full response.
- Non-streaming: build the complete response directly.

### Mapper conventions

- `convert_messages(messages) -> provider format` — converts `Sequence[BaseEvent]` to the provider's message list.
- `tool_to_api(tool) -> dict` — converts a `ToolSchema` to the provider's tool definition. Use `isinstance()` checks; unsupported tools fall through to `raise UnsupportedToolError(t.type, "provider")`.
- `response_proto_to_*(schema)` — converts `ResponseProto` to the provider's structured-output format. Use `_ensure_additional_properties_false()` where the provider requires it.
- `normalize_usage(usage) -> Usage` — maps provider-specific usage keys to the normalised `Usage` dataclass.

### Adding a new provider

1. Create `ag2/config/{provider}/` with `config.py`, `{provider}_client.py`, and `mappers.py`.
2. Register the config in `ag2/config/__init__.py`: import inside a `try/except ImportError` block and add a `_missing_optional_dependency_config` fallback.
3. Add the config to `__all__`.
4. Add mapper tests under `test/config/{provider}/`