Skip to content

PROVIDER_CONTRACT.md

Latest commit

 

History

History
277 lines (196 loc) · 8.71 KB

PROVIDER_CONTRACT.md

File metadata and controls

277 lines (196 loc) · 8.71 KB
contract_type module_specification
module_type provider
contract_version 1.0.0
last_modified 2025-01-29
related_files
path relationship
../../python/amplifier_core/interfaces.py#Provider
protocol_definition
path relationship
../../python/amplifier_core/message_models.py
request_response_models
path relationship
../../python/amplifier_core/content_models.py
event_content_types
path relationship
../../python/amplifier_core/models.py#ProviderInfo
metadata_models
path relationship
../specs/PROVIDER_SPECIFICATION.md
detailed_spec
path relationship
../specs/MOUNT_PLAN_SPECIFICATION.md
configuration
path relationship
../specs/CONTRIBUTION_CHANNELS.md
observability
path relationship
../../python/amplifier_core/testing.py
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 → class Provider(Protocol)

@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)

ModelInfo Extensions (Model Class Routing)

The list_models() method returns list[ModelInfo]. Beyond the required fields (id, display_name, context_window, max_output_tokens), ModelInfo supports optional extension fields for model class routing and cost-aware selection:

Field Type Default Description
cost_per_input_token float | None None Cost per input token in USD (e.g., 3e-6 for $3/MTok)
cost_per_output_token float | None None Cost per output token in USD
metadata dict[str, Any] {} Extensible metadata bag for cost tier, model class, provider-specific tags

Cost Fields

Providers SHOULD populate cost_per_input_token and cost_per_output_token when pricing information is available. These enable cost-aware model selection and budget tracking.

Metadata: cost_tier

Providers SHOULD set metadata["cost_tier"] to one of the well-known cost tier strings:

Tier Description
free No-cost models (local, free-tier)
low Budget-friendly models (e.g., Haiku-class)
medium Standard pricing (e.g., Sonnet-class)
high Premium pricing (e.g., Opus-class)
extreme Highest-cost models (e.g., deep research)

Capabilities

Providers SHOULD populate the capabilities list using well-known constants from amplifier_core.capabilities. See the Capabilities Taxonomy in the Provider Specification for the full list.

Backward Compatibility

All extension fields are optional with sensible defaults. Existing providers that do not populate these fields continue to work unchanged — they simply won't participate in cost-aware or capability-based routing.

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.

llm:response Event — usage Payload Schema

Providers MUST emit llm:response with the following usage payload. Key names are normative — derived from the kernel Usage struct (crates/amplifier-core/src/messages.rs):

Key Type Required Description
input_tokens int MUST Total input tokens — gross total (fresh + cache_read combined)
output_tokens int MUST Total output tokens generated
cache_read_tokens int SHOULD (when non-zero) Tokens served from prompt cache
cache_write_tokens int SHOULD (when non-zero) Tokens written to prompt cache (billed on top of gross)

DO NOT use: "input", "output", "input_tokens_used", "completion_tokens", or any other variant. These break consumers silently.

total_tokens is not included in the event payload. Consumers that need it can derive it as input_tokens + output_tokens from the event usage dict.

Reference emit pattern (build ChatResponse before emitting):

chat_response = self._convert_to_chat_response(response)  # build first

event_usage: dict[str, Any] = {
    "input_tokens": chat_response.usage.input_tokens,
    "output_tokens": chat_response.usage.output_tokens,
}
if chat_response.usage.cache_read_tokens is not None:
    event_usage["cache_read_tokens"] = chat_response.usage.cache_read_tokens
if chat_response.usage.cache_write_tokens is not None:
    event_usage["cache_write_tokens"] = chat_response.usage.cache_write_tokens

await coordinator.hooks.emit("llm:response", {
    "provider": self.name,
    "model": model,
    "status": "ok",
    "duration_ms": elapsed_ms,
    "usage": event_usage,
})

return chat_response  # return the already-built response

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