Module Development

Module Contracts

Start here for building Amplifier modules.

This directory contains the authoritative guidance for building each type of Amplifier module. Each contract document explains:

1. What it is - Purpose and responsibilities
2. Protocol reference - Link to interfaces.py with exact line numbers
3. Entry point pattern - How modules are discovered and loaded
4. Configuration - Mount Plan integration
5. Canonical example - Reference implementation
6. Validation - How to verify your module works

Module Types

Quick Start Pattern

All modules follow this pattern:

# 1. Implement the Protocol from interfaces.py
class MyModule:
    # ... implement required methods
    pass

# 2. Provide mount() function
async def mount(coordinator, config):
    """Initialize and register module."""
    instance = MyModule(config)
    await coordinator.mount("category", instance, name="my-module")
    return instance  # or cleanup function

# 3. Register entry point in pyproject.toml
# [project.entry-points."amplifier.modules"]
# my-module = "my_package:mount"

Source of Truth

Protocols are in code, not docs:

• Protocol definitions: amplifier_core/interfaces.py
• Data models: amplifier_core/models.py
• Message models: amplifier_core/message_models.py (Pydantic models for request/response envelopes)
• Content models: amplifier_core/content_models.py (dataclass types for events and streaming)

These contract documents provide guidance that code cannot express. Always read the code docstrings first.

Related Documentation

• MOUNT_PLAN_SPECIFICATION.md - Configuration contract
• MODULE_SOURCE_PROTOCOL.md - Module loading mechanism
• CONTRIBUTION_CHANNELS.md - Module contribution pattern
• DESIGN_PHILOSOPHY.md - Kernel design principles

Validation

Verify your module before release:

# Structural validation
amplifier module validate ./my-module

See individual contract documents for type-specific validation requirements.

For ecosystem overview: amplifier

contract_type: module_specification module_type: tool contract_version: 1.0.0 last_modified: 2025-01-29 related_files:

• path: amplifier_core/interfaces.py#Tool relationship: protocol_definition lines: 121-146
• path: amplifier_core/models.py#ToolResult relationship: result_model
• path: amplifier_core/message_models.py#ToolCall relationship: invocation_model
• path: ../specs/MOUNT_PLAN_SPECIFICATION.md relationship: configuration
• path: amplifier_core/testing.py#MockTool relationship: test_utilities canonical_example: https://github.com/microsoft/amplifier-module-tool-filesystem

Tool Contract

Tools provide capabilities that agents can invoke during execution.

Purpose

Tools extend agent capabilities beyond pure conversation:

• Filesystem operations - Read, write, edit files
• Command execution - Run shell commands
• Web access - Fetch URLs, search
• Task delegation - Spawn sub-agents
• Custom capabilities - Domain-specific operations

Protocol Definition

Source: amplifier_core/interfaces.py lines 121-146

@runtime_checkable
class Tool(Protocol):
    @property
    def name(self) -> str:
        """Tool name for invocation."""
        ...

    @property
    def description(self) -> str:
        """Human-readable tool description."""
        ...

    async def execute(self, input: dict[str, Any]) -> ToolResult:
        """
        Execute tool with given input.

        Args:
            input: Tool-specific input parameters

        Returns:
            Tool execution result
        """
        ...

Data Models

ToolCall (Input)

Source: amplifier_core/message_models.py

class ToolCall(BaseModel):
    id: str                    # Unique ID for correlation
    name: str                  # Tool name to invoke
    arguments: dict[str, Any]  # Tool-specific parameters

ToolResult (Output)

Source: amplifier_core/models.py

class ToolResult(BaseModel):
    success: bool = True              # Whether execution succeeded
    output: Any | None = None         # Tool output (typically str or dict)
    error: dict[str, Any] | None = None  # Error details if failed

Entry Point Pattern

mount() Function

async def mount(coordinator: ModuleCoordinator, config: dict) -> Tool | Callable | None:
    """
    Initialize and register tool.

    Returns:
        - Tool instance
        - Cleanup callable (for resource cleanup)
        - None for graceful degradation
    """
    tool = MyTool(config=config)
    await coordinator.mount("tools", tool, name="my-tool")
    return tool

pyproject.toml

[project.entry-points."amplifier.modules"]
my-tool = "my_tool:mount"

Implementation Requirements

Name and Description

Tools must provide clear identification:

class MyTool:
    @property
    def name(self) -> str:
        return "my_tool"  # Used for invocation

    @property
    def description(self) -> str:
        return "Performs specific action with given parameters."

Best practices:

• name: Short, snake_case, unique across mounted tools
• description: Clear explanation of what the tool does and expects

execute() Method

Handle inputs and return structured results:

async def execute(self, input: dict[str, Any]) -> ToolResult:
    try:
        # Validate input
        required_param = input.get("required_param")
        if not required_param:
            return ToolResult(
                success=False,
                error={"message": "required_param is required"}
            )

        # Do the work
        result = await self._do_work(required_param)

        return ToolResult(
            success=True,
            output=result
        )

    except Exception as e:
        return ToolResult(
            success=False,
            error={"message": str(e), "type": type(e).__name__}
        )

Tool Schema (Optional but Recommended)

Provide JSON schema for input validation:

def get_schema(self) -> dict:
    """Return JSON schema for tool input."""
    return {
        "type": "object",
        "properties": {
            "required_param": {
                "type": "string",
                "description": "Description of parameter"
            },
            "optional_param": {
                "type": "integer",
                "default": 10
            }
        },
        "required": ["required_param"]
    }

Configuration

Tools receive configuration via Mount Plan:

tools:
  - module: my-tool
    source: git+https://github.com/org/my-tool@main
    config:
      max_size: 1048576
      allowed_paths:
        - /home/user/projects

See MOUNT_PLAN_SPECIFICATION.md for full schema.

Observability

Register lifecycle events:

coordinator.register_contributor(
    "observability.events",
    "my-tool",
    lambda: ["my-tool:started", "my-tool:completed", "my-tool:error"]
)

Standard tool events emitted by orchestrators:

• tool:pre - Before tool execution
• tool:post - After successful execution
• tool:error - On execution failure

Canonical Example

Reference implementation: amplifier-module-tool-filesystem

Study this module for:

• Tool protocol implementation
• Input validation patterns
• Error handling and result formatting
• Configuration integration

Additional examples:

• amplifier-module-tool-bash - Command execution
• amplifier-module-tool-web - Web access

Validation Checklist

Required

• Implements Tool protocol (name, description, execute)
• mount() function with entry point in pyproject.toml
• Returns ToolResult from execute()
• Handles errors gracefully (returns success=False, doesn't crash)

Recommended

• Provides JSON schema via get_schema()
• Validates input before processing
• Logs operations at appropriate levels
• Registers observability events

Testing

Use test utilities from amplifier_core/testing.py:

from amplifier_core.testing import TestCoordinator, MockTool

@pytest.mark.asyncio
async def test_tool_execution():
    tool = MyTool(config={})

    result = await tool.execute({
        "required_param": "value"
    })

    assert result.success
    assert result.error is None

MockTool for Testing Orchestrators

from amplifier_core.testing import MockTool

mock_tool = MockTool(
    name="test_tool",
    description="Test tool",
    return_value="mock result"
)

# After use
assert mock_tool.call_count == 1
assert mock_tool.last_input == {...}

Quick Validation Command

# Structural validation
amplifier module validate ./my-tool --type tool

Related: README.md | HOOK_CONTRACT.md

contract_type: module_specification module_type: provider contract_version: 1.0.0 last_modified: 2025-01-29 related_files:

• path: amplifier_core/interfaces.py#Provider relationship: protocol_definition lines: 54-119
• path: amplifier_core/message_models.py relationship: request_response_models
• path: amplifier_core/content_models.py relationship: event_content_types
• path: amplifier_core/models.py#ProviderInfo relationship: metadata_models
• path: ../specs/PROVIDER_SPECIFICATION.md relationship: detailed_spec
• path: ../specs/MOUNT_PLAN_SPECIFICATION.md relationship: configuration
• path: ../specs/CONTRIBUTION_CHANNELS.md relationship: observability
• path: amplifier_core/testing.py relationship: test_utilities canonical_example: https://github.com/microsoft/amplifier-module-provider-anthropic

Provider Contract

Providers translate between Amplifier's unified message format and vendor-specific LLM APIs.

Detailed Specification

See PROVIDER_SPECIFICATION.md for complete implementation guidance including:

• Protocol summary and method signatures
• Content block preservation requirements
• Role conversion patterns
• Auto-continuation handling
• Debug levels and observability

This contract document provides the quick-reference essentials. The specification contains the full details.

Protocol Definition

Source: amplifier_core/interfaces.py lines 54-119

@runtime_checkable
class Provider(Protocol):
    @property
    def name(self) -> str: ...

    def get_info(self) -> ProviderInfo: ...

    async def list_models(self) -> list[ModelInfo]: ...

    async def complete(self, request: ChatRequest, **kwargs) -> ChatResponse: ...

    def parse_tool_calls(self, response: ChatResponse) -> list[ToolCall]: ...

Note: ToolCall is from amplifier_core.message_models (see REQUEST_ENVELOPE_V1 for details)

Entry Point Pattern

mount() Function

async def mount(coordinator: ModuleCoordinator, config: dict) -> Provider | Callable | None:
    """
    Initialize and return provider instance.

    Returns:
        - Provider instance (registered automatically)
        - Cleanup callable (for resource cleanup on unmount)
        - None for graceful degradation (e.g., missing API key)
    """
    api_key = config.get("api_key") or os.environ.get("MY_API_KEY")
    if not api_key:
        logger.warning("No API key - provider not mounted")
        return None

    provider = MyProvider(api_key=api_key, config=config)
    await coordinator.mount("providers", provider, name="my-provider")

    async def cleanup():
        await provider.client.close()

    return cleanup

pyproject.toml

[project.entry-points."amplifier.modules"]
my-provider = "my_provider:mount"

Configuration

Providers receive configuration via Mount Plan:

providers:
  - module: my-provider
    source: git+https://github.com/org/my-provider@main
    config:
      api_key: "${MY_API_KEY}"
      default_model: model-v1
      debug: true

See MOUNT_PLAN_SPECIFICATION.md for full schema.

Observability

Register custom events via contribution channels:

coordinator.register_contributor(
    "observability.events",
    "my-provider",
    lambda: ["my-provider:rate_limit", "my-provider:retry"]
)

See CONTRIBUTION_CHANNELS.md for the pattern.

Canonical Example

Reference implementation: amplifier-module-provider-anthropic

Study this module for:

• Complete Provider protocol implementation
• Content block handling patterns
• Configuration and credential management
• Debug logging integration

Validation Checklist

Required

• Implements all 5 Provider protocol methods
• mount() function with entry point in pyproject.toml
• Preserves all content block types (especially signature in ThinkingBlock)
• Reports Usage (input/output/total tokens)
• Returns ChatResponse from complete()

Recommended

• Graceful degradation on missing config (return None from mount)
• Validates tool call/result sequences
• Supports debug configuration flags
• Registers cleanup function for resource management
• Registers observability events via contribution channels

Testing

Use test utilities from amplifier_core/testing.py:

from amplifier_core.testing import TestCoordinator, create_test_coordinator

@pytest.mark.asyncio
async def test_provider_mount():
    coordinator = create_test_coordinator()
    cleanup = await mount(coordinator, {"api_key": "test-key"})

    assert "my-provider" in coordinator.get_mounted("providers")

    if cleanup:
        await cleanup()

Quick Validation Command

# Structural validation
amplifier module validate ./my-provider --type provider

Related: PROVIDER_SPECIFICATION.md | README.md

contract_type: module_specification module_type: hook contract_version: 1.0.0 last_modified: 2025-01-29 related_files:

• path: amplifier_core/interfaces.py#HookHandler relationship: protocol_definition lines: 205-220
• path: amplifier_core/models.py#HookResult relationship: result_model
• path: ../HOOKS_API.md relationship: detailed_api
• path: ../specs/MOUNT_PLAN_SPECIFICATION.md relationship: configuration
• path: ../specs/CONTRIBUTION_CHANNELS.md relationship: observability
• path: amplifier_core/testing.py#EventRecorder relationship: test_utilities canonical_example: https://github.com/microsoft/amplifier-module-hooks-logging

Hook Contract

Hooks observe, validate, and control agent lifecycle events.

Purpose

Hooks enable:

• Observation - Logging, metrics, audit trails
• Validation - Security checks, input validation
• Feedback injection - Automated correction loops
• Approval gates - Dynamic permission requests
• Output control - Clean user experience

Detailed API Reference

See HOOKS_API.md for complete documentation including:

• HookResult actions and fields
• Registration patterns
• Common patterns with examples
• Best practices

This contract provides the essentials. The API reference contains full details.

Protocol Definition

Source: amplifier_core/interfaces.py lines 205-220

@runtime_checkable
class HookHandler(Protocol):
    async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
        """
        Handle a lifecycle event.

        Args:
            event: Event name (e.g., "tool:pre", "session:start")
            data: Event-specific data

        Returns:
            HookResult indicating action to take
        """
        ...

HookResult Actions

Source: amplifier_core/models.py

from amplifier_core.models import HookResult

# Simple observation
HookResult(action="continue")

# Block with reason
HookResult(action="deny", reason="Access denied")

# Inject feedback
HookResult(
    action="inject_context",
    context_injection="Found 3 linting errors...",
    user_message="Linting issues detected"
)

# Request approval
HookResult(
    action="ask_user",
    approval_prompt="Allow write to production file?",
    approval_default="deny"
)

Entry Point Pattern

mount() Function

async def mount(coordinator: ModuleCoordinator, config: dict) -> Callable | None:
    """
    Initialize and register hook handlers.

    Returns:
        Cleanup callable to unregister handlers
    """
    handlers = []

    # Register handlers for specific events
    handlers.append(
        coordinator.hooks.register("tool:pre", my_validation_hook, priority=10)
    )
    handlers.append(
        coordinator.hooks.register("tool:post", my_feedback_hook, priority=20)
    )

    # Return cleanup function
    def cleanup():
        for unregister in handlers:
            unregister()

    return cleanup

pyproject.toml

[project.entry-points."amplifier.modules"]
my-hook = "my_hook:mount"

Event Registration

Register handlers during mount():

from amplifier_core.hooks import HookRegistry

# Get registry from coordinator
registry: HookRegistry = coordinator.hooks

# Register with priority (lower = earlier)
unregister = registry.register(
    event="tool:post",
    handler=my_handler,
    priority=10,
    name="my_handler"
)

# Later: unregister()

Common Events

Configuration

Hooks receive configuration via Mount Plan:

hooks:
  - module: my-hook
    source: git+https://github.com/org/my-hook@main
    config:
      enabled_events:
        - "tool:pre"
        - "tool:post"
      log_level: "info"

See MOUNT_PLAN_SPECIFICATION.md for full schema.

Observability

Register custom events your hook emits:

coordinator.register_contributor(
    "observability.events",
    "my-hook",
    lambda: ["my-hook:validation_failed", "my-hook:approved"]
)

See CONTRIBUTION_CHANNELS.md for the pattern.

Canonical Example

Reference implementation: amplifier-module-hooks-logging

Study this module for:

• Hook registration patterns
• Event handling
• Configuration integration
• Observability contribution

Additional examples:

• amplifier-module-hooks-approval - Approval gates
• amplifier-module-hooks-redaction - Security redaction

Validation Checklist

Required

• Handler implements async def __call__(event, data) -> HookResult
• mount() function with entry point in pyproject.toml
• Returns valid HookResult for all code paths
• Handles exceptions gracefully (don't crash kernel)

Recommended

• Register cleanup function to unregister handlers
• Use appropriate priority (10-90, lower = earlier)
• Log handler registration for debugging
• Support configuration for enabled events
• Register custom events via contribution channels

Testing

Use test utilities from amplifier_core/testing.py:

from amplifier_core.testing import TestCoordinator, EventRecorder
from amplifier_core.models import HookResult

@pytest.mark.asyncio
async def test_hook_handler():
    # Test handler directly
    result = await my_validation_hook("tool:pre", {
        "tool_name": "Write",
        "tool_input": {"file_path": "/etc/passwd"}
    })

    assert result.action == "deny"
    assert "denied" in result.reason.lower()

@pytest.mark.asyncio
async def test_hook_registration():
    coordinator = TestCoordinator()
    cleanup = await mount(coordinator, {})

    # Verify handlers registered
    # ... test event emission

    cleanup()

EventRecorder for Testing

from amplifier_core.testing import EventRecorder

recorder = EventRecorder()

# Use in tests
await recorder.record("tool:pre", {"tool_name": "Write"})

# Assert
events = recorder.get_events()
assert len(events) == 1
assert events[0][0] == "tool:pre"  # events are (event_name, data) tuples

Quick Validation Command

# Structural validation
amplifier module validate ./my-hook --type hook

Related: HOOKS_API.md | README.md

contract_type: module_specification module_type: orchestrator contract_version: 1.0.0 last_modified: 2025-01-29 related_files:

• path: amplifier_core/interfaces.py#Orchestrator relationship: protocol_definition lines: 26-52
• path: amplifier_core/content_models.py relationship: event_content_types
• path: ../specs/MOUNT_PLAN_SPECIFICATION.md relationship: configuration
• path: ../specs/CONTRIBUTION_CHANNELS.md relationship: observability
• path: amplifier_core/testing.py#ScriptedOrchestrator relationship: test_utilities canonical_example: https://github.com/microsoft/amplifier-module-loop-basic

Orchestrator Contract

Orchestrators implement the agent execution loop strategy.

Purpose

Orchestrators control how agents execute:

• Basic loops - Simple prompt → response → tool → response cycles
• Streaming - Real-time response delivery
• Event-driven - Complex multi-step workflows
• Custom strategies - Domain-specific execution patterns

Key principle: The orchestrator is policy, not mechanism. Swap orchestrators to change agent behavior without modifying the kernel.

Protocol Definition

Source: amplifier_core/interfaces.py lines 26-52

@runtime_checkable
class Orchestrator(Protocol):
    async def execute(
        self,
        prompt: str,
        context: ContextManager,
        providers: dict[str, Provider],
        tools: dict[str, Tool],
        hooks: HookRegistry,
    ) -> str:
        """
        Execute the agent loop with given prompt.

        Args:
            prompt: User input prompt
            context: Context manager for conversation state
            providers: Available LLM providers (keyed by name)
            tools: Available tools (keyed by name)
            hooks: Hook registry for lifecycle events

        Returns:
            Final response string
        """
        ...

Execution Flow

A typical orchestrator implements this flow:

User Prompt
    ↓
Add to Context
    ↓
┌─────────────────────────────────────┐
│  LOOP until response has no tools   │
│                                     │
│  1. emit("provider:request")        │
│  2. provider.complete(messages)     │
│  3. emit("provider:response")       │
│  4. Add response to context         │
│                                     │
│  If tool_calls:                     │
│    for each tool_call:              │
│      5. emit("tool:pre")            │
│      6. tool.execute(input)         │
│      7. emit("tool:post")           │
│      8. Add result to context       │
│                                     │
│  Continue loop...                   │
└─────────────────────────────────────┘
    ↓
Return final text response

Entry Point Pattern

mount() Function

async def mount(coordinator: ModuleCoordinator, config: dict) -> Orchestrator | Callable | None:
    """
    Initialize and return orchestrator instance.

    Returns:
        - Orchestrator instance
        - Cleanup callable
        - None for graceful degradation
    """
    orchestrator = MyOrchestrator(config=config)
    await coordinator.mount("session", orchestrator, name="orchestrator")
    return orchestrator

pyproject.toml

[project.entry-points."amplifier.modules"]
my-orchestrator = "my_orchestrator:mount"

Implementation Requirements

Event Emission

Orchestrators must emit lifecycle events for observability:

async def execute(self, prompt, context, providers, tools, hooks):
    # Before LLM call
    await hooks.emit("provider:request", {
        "provider": provider_name,
        "messages": messages,
        "model": model_name
    })

    response = await provider.complete(request)

    # After LLM call
    await hooks.emit("provider:response", {
        "provider": provider_name,
        "response": response,
        "usage": response.usage
    })

    # Before tool execution
    await hooks.emit("tool:pre", {
        "tool_name": tool_call.name,
        "tool_input": tool_call.input
    })

    result = await tool.execute(tool_call.input)

    # After tool execution
    await hooks.emit("tool:post", {
        "tool_name": tool_call.name,
        "tool_input": tool_call.input,
        "tool_result": result
    })

Hook Processing

Handle HookResult actions:

# Before tool execution
pre_result = await hooks.emit("tool:pre", data)

if pre_result.action == "deny":
    # Don't execute tool
    return ToolResult(is_error=True, output=pre_result.reason)

if pre_result.action == "modify":
    # Use modified data
    data = pre_result.data

if pre_result.action == "inject_context":
    # Add feedback to context
    await context.add_message({
        "role": pre_result.context_injection_role,
        "content": pre_result.context_injection
    })

if pre_result.action == "ask_user":
    # Request approval (requires approval provider)
    approved = await request_approval(pre_result)
    if not approved:
        return ToolResult(is_error=True, output="User denied")

Context Management

Manage conversation state:

# Add user message
await context.add_message({"role": "user", "content": prompt})

# Add assistant response
await context.add_message({"role": "assistant", "content": response.content})

# Add tool result
await context.add_message({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": result.output
})

# Get messages for LLM request (context handles compaction internally)
messages = await context.get_messages_for_request()

Provider Selection

Handle multiple providers:

# Get default or configured provider
provider_name = config.get("default_provider", list(providers.keys())[0])
provider = providers[provider_name]

# Or allow per-request provider selection
provider_name = request_options.get("provider", default_provider_name)

Configuration

Orchestrators receive configuration via Mount Plan:

session:
  orchestrator: my-orchestrator
  context: context-simple

# Orchestrator-specific config can be passed via providers/tools config

See MOUNT_PLAN_SPECIFICATION.md for full schema.

Observability

Register custom events your orchestrator emits:

coordinator.register_contributor(
    "observability.events",
    "my-orchestrator",
    lambda: [
        "my-orchestrator:loop_started",
        "my-orchestrator:loop_iteration",
        "my-orchestrator:loop_completed"
    ]
)

See CONTRIBUTION_CHANNELS.md for the pattern.

Canonical Example

Reference implementation: amplifier-module-loop-basic

Study this module for:

• Complete execute() implementation
• Event emission patterns
• Hook result handling
• Context management

Additional examples:

• amplifier-module-loop-streaming - Real-time streaming
• amplifier-module-loop-events - Event-driven patterns

Validation Checklist

Required

• Implements execute(prompt, context, providers, tools, hooks) -> str
• mount() function with entry point in pyproject.toml
• Emits standard events (provider:request/response, tool:pre/post)
• Handles HookResult actions appropriately
• Manages context (add messages, check compaction)

Recommended

• Supports multiple providers
• Implements max iterations limit (prevent infinite loops)
• Handles provider errors gracefully
• Registers custom observability events
• Supports streaming via async generators

Testing

Use test utilities from amplifier_core/testing.py:

from amplifier_core.testing import (
    TestCoordinator,
    MockTool,
    MockContextManager,
    ScriptedOrchestrator,
    EventRecorder
)

@pytest.mark.asyncio
async def test_orchestrator_basic():
    orchestrator = MyOrchestrator(config={})
    context = MockContextManager()
    providers = {"test": MockProvider()}
    tools = {"test_tool": MockTool()}
    hooks = HookRegistry()

    result = await orchestrator.execute(
        prompt="Test prompt",
        context=context,
        providers=providers,
        tools=tools,
        hooks=hooks
    )

    assert isinstance(result, str)
    assert len(context.messages) > 0

ScriptedOrchestrator for Testing

from amplifier_core.testing import ScriptedOrchestrator

# For testing components that use orchestrators
orchestrator = ScriptedOrchestrator(responses=["Response 1", "Response 2"])

result = await orchestrator.execute(...)
assert result == "Response 1"

Quick Validation Command

# Structural validation
amplifier module validate ./my-orchestrator --type orchestrator

Related: README.md | CONTEXT_CONTRACT.md

contract_type: module_specification module_type: context contract_version: 2.1.0 last_modified: 2026-01-01 related_files:

• path: amplifier_core/interfaces.py#ContextManager relationship: protocol_definition lines: 148-180
• path: ../specs/MOUNT_PLAN_SPECIFICATION.md relationship: configuration
• path: ../specs/CONTRIBUTION_CHANNELS.md relationship: observability
• path: amplifier_core/testing.py#MockContextManager relationship: test_utilities canonical_example: https://github.com/microsoft/amplifier-module-context-simple

Context Contract

Context managers handle conversation memory and message storage.

Purpose

Context managers control what the agent remembers:

• Message storage - Store conversation history
• Request preparation - Return messages that fit within token limits
• Persistence - Optionally persist across sessions
• Memory strategies - Implement various memory patterns

Key principle: The context manager owns policy for memory. The orchestrator asks for messages; the context manager decides how to fit them within limits. Swap context managers to change memory behavior without modifying orchestrators.

Mechanism vs Policy: Orchestrators provide the mechanism (request messages, make LLM calls). Context managers provide the policy (what to return, when to compact, how to fit within limits).

Protocol Definition

Source: amplifier_core/interfaces.py lines 148-180

@runtime_checkable
class ContextManager(Protocol):
    async def add_message(self, message: dict[str, Any]) -> None:
        """Add a message to the context."""
        ...

    async def get_messages_for_request(
        self,
        token_budget: int | None = None,
        provider: Any | None = None,
    ) -> list[dict[str, Any]]:
        """
        Get messages ready for an LLM request.

        The context manager handles any compaction needed internally.
        Returns messages that fit within the token budget.

        Args:
            token_budget: Optional explicit token limit (deprecated, prefer provider).
            provider: Optional provider instance for dynamic budget calculation.
                If provided, budget = context_window - max_output_tokens - safety_margin.

        Returns:
            Messages ready for LLM request, compacted if necessary.
        """
        ...

    async def get_messages(self) -> list[dict[str, Any]]:
        """Get all messages (raw, uncompacted) for transcripts/debugging."""
        ...

    async def set_messages(self, messages: list[dict[str, Any]]) -> None:
        """Set messages directly (for session resume)."""
        ...

    async def clear(self) -> None:
        """Clear all messages."""
        ...

Message Format

Messages follow a standard structure:

# User message
{
    "role": "user",
    "content": "User's input text"
}

# Assistant message
{
    "role": "assistant",
    "content": "Assistant's response"
}

# Assistant message with tool calls
{
    "role": "assistant",
    "content": None,
    "tool_calls": [
        {
            "id": "call_123",
            "type": "function",
            "function": {"name": "read_file", "arguments": "{...}"}
        }
    ]
}

# System message
{
    "role": "system",
    "content": "System instructions"
}

# Tool result
{
    "role": "tool",
    "tool_call_id": "call_123",
    "content": "Tool output"
}

Entry Point Pattern

mount() Function

async def mount(coordinator: ModuleCoordinator, config: dict) -> ContextManager | Callable | None:
    """
    Initialize and return context manager instance.

    Returns:
        - ContextManager instance
        - Cleanup callable
        - None for graceful degradation
    """
    context = MyContextManager(
        max_tokens=config.get("max_tokens", 100000),
        compaction_threshold=config.get("compaction_threshold", 0.8)
    )
    await coordinator.mount("session", context, name="context")
    return context

pyproject.toml

[project.entry-points."amplifier.modules"]
my-context = "my_context:mount"

Implementation Requirements

add_message()

Store messages with proper validation:

async def add_message(self, message: dict[str, Any]) -> None:
    """Add a message to the context."""
    # Validate required fields
    if "role" not in message:
        raise ValueError("Message must have 'role' field")

    # Store message
    self._messages.append(message)

    # Track token count (approximate)
    self._token_count += self._estimate_tokens(message)

get_messages_for_request()

Return messages ready for LLM request, handling compaction internally:

async def get_messages_for_request(
    self,
    token_budget: int | None = None,
    provider: Any | None = None,
) -> list[dict[str, Any]]:
    """
    Get messages ready for an LLM request.

    Handles compaction internally if needed. Orchestrators call this
    before every LLM request and trust the context manager to return
    messages that fit within limits.

    Args:
        token_budget: Optional explicit token limit (deprecated, prefer provider).
        provider: Optional provider instance for dynamic budget calculation.
            If provided, budget = context_window - max_output_tokens - safety_margin.
    """
    budget = self._calculate_budget(token_budget, provider)

    # Check if compaction needed
    if self._token_count > (budget * self._compaction_threshold):
        await self._compact_internal()

    return list(self._messages)  # Return copy to prevent mutation

def _calculate_budget(self, token_budget: int | None, provider: Any | None) -> int:
    """Calculate effective token budget from provider or fallback to config."""
    # Explicit budget takes precedence (for backward compatibility)
    if token_budget is not None:
        return token_budget

    # Try provider-based dynamic budget
    if provider is not None:
        try:
            info = provider.get_info()
            defaults = info.defaults or {}
            context_window = defaults.get("context_window")
            max_output_tokens = defaults.get("max_output_tokens")

            if context_window and max_output_tokens:
                safety_margin = 1000  # Buffer to avoid hitting hard limits
                return context_window - max_output_tokens - safety_margin
        except Exception:
            pass  # Fall back to configured max_tokens

    return self._max_tokens

get_messages()

Return all messages for transcripts/debugging (no compaction):

async def get_messages(self) -> list[dict[str, Any]]:
    """Get all messages (raw, uncompacted) for transcripts/debugging."""
    return list(self._messages)  # Return copy to prevent mutation

set_messages()

Set messages directly for session resume:

async def set_messages(self, messages: list[dict[str, Any]]) -> None:
    """Set messages directly (for session resume)."""
    self._messages = list(messages)
    self._token_count = sum(self._estimate_tokens(m) for m in self._messages)

File-Based Context Managers - Special Behavior:

For context managers with persistent file storage (like context-persistent), the behav

…(truncated)