Extensions

Extensions

Extensions are add-on plugins that react to runtime events without replacing any core functionality. The core emits lifecycle hooks - extensions subscribe to the ones they care about. The core never imports extensions and has no idea who is listening.

Use Cases

Extensions are the right choice when you want to observe or react to what the runtime is doing, without changing how it works:

• Notifications - Send a Telegram message when a job completes, a payment settles, or the wallet balance drops below a threshold. The built-in webhook-notifier plugin works this way.
• Monitoring and metrics export - Push job completion rates, error counts, and earnings data to Prometheus, Datadog, or a custom dashboard.
• Audit logging - Write a structured audit trail of every job, payment, and agent decision to an external log store for compliance.
• Custom alerting - Page the provider when the agent hits repeated errors on a platform, or when a capability’s spec drifts.
• External integrations - Post to Slack when the adapter registers on a new platform, or update a CRM when a job completes.

The key distinction from tool plugins: extensions don’t give the agent new abilities. The agent doesn’t call extensions - the runtime fires events and extensions react. Extensions are for the provider’s operational awareness, not for the agent’s decision-making.

Available Hooks

Hook	Emitted When
on_job_complete	A job finishes successfully
on_job_failed	A job fails
on_low_balance	Wallet balance drops below configured threshold
on_platform_registered	The agent registers on a new platform
on_agent_error	The agent encounters an error during its loop
on_capability_drift	A spec change is detected in the capability source

Custom string hooks are also supported - extensions can subscribe to any string event the runtime emits.

Interface

Every extension implements the Extension contract from agent-adapter-contracts:

from agent_adapter_contracts.extensions import Extension, RuntimeEvent
from agent_adapter_contracts.runtime import RuntimeAPI

class MyExtension(Extension):
    @property
    def name(self) -> str:
        return "my-notifier"

    @property
    def hooks(self) -> list:
        return [
            RuntimeEvent.ON_JOB_COMPLETE,
            RuntimeEvent.ON_JOB_FAILED,
            RuntimeEvent.ON_LOW_BALANCE,
        ]

    async def initialize(self, runtime: RuntimeAPI) -> None:
        """Called once at startup. Set up connections, load config."""
        self.webhook_url = await runtime.secrets.retrieve("notifier", "webhook_url")

    async def shutdown(self) -> None:
        """Called when the runtime stops. Clean up resources."""
        pass

Note

Extensions don’t implement individual hook handler methods on the ABC. The ExtensionRegistry matches hooks to extensions via the hooks list and dispatches events. Your extension receives events through the registry’s emit mechanism.

How It Works

The runtime’s ExtensionRegistry is simple:

1. Extensions register themselves with the hooks they care about
2. When the runtime reaches a lifecycle point (job completes, balance drops, etc.), it calls registry.emit(hook, payload)
3. The registry fans out to all extensions subscribed to that hook
4. Failures in one extension don’t affect others - Promise.allSettled semantics

# Inside the job engine - core has no idea who is listening
await self.extensions.emit("on_job_complete", job)

Configuration

extensions:
  - id: "webhook-notifier"
    config:
      webhook_url: "${WEBHOOK_URL}"
      events: ["on_job_complete", "on_job_failed"]

Built-in Extension

webhook-notifier

Sends HTTP POST requests to a configured webhook URL when subscribed events fire. Useful for connecting to Slack, Discord, PagerDuty, or any webhook-compatible service.

Publishing

[project.entry-points."agent_adapter.extensions"]
my_extension = "my_extension_plugin:MyExtension"