Writing an integration¶

Available · agent_adapter.integrations.base · .discovery

An integration turns registry tools into one framework's native tools. You can ship it in your own package: register it under the agent_adapter.integrations entry-point group and the adapter finds it, with no change to the adapter.

sequenceDiagram
  participant F as Framework
  participant T as Exported tool
  participant I as Integration
  participant R as ToolRunner (Executor)
  participant S as ToolSpec
  F->>T: native call (id, arguments)
  T->>I: bind_context(native)
  I-->>T: ToolContext(call_id, extras)
  T->>R: run_spec(spec, args, ctx, call_id=…)
  R->>S: validate, hooks, approval, timeout, retries, cache, ledger
  S-->>R: value or error
  R-->>T: value, or raises
  T-->>F: native result (model-visible error when the framework has no error channel)

The protocol¶

Member	Meaning
`name`	Canonical target name. Must equal the entry-point name
`requires`	Import names that must be importable (`("google.genai",)`)
`extra`	Pip extra named in the `MissingDependency` message
`export_tool(spec, *, executor=None)`	The framework's native tool for one `ToolSpec`
`bind_context(native)`	A `ToolContext` for one native call; carries the native call id
`bind_approval(port)`	A native approval gate, or `None` when the executor's port decides
`compile_flow(flow, , executor=None, *options)`	Optional (`FlowCompiler`): a flow as the framework's graph

IntegrationBase gives you the rest: export_tools(registry) (honours targets), bind_context (reads call_id_fields from the native call), bind_approval (None), require(), and _run_bound(native, spec, args, runner) for frameworks without a tool-error channel (bind → run_spec(..., call_id=ctx.call_id) → model-visible errors).

1. Write the integration¶

The demo plugin below drives a tiny framework whose tools are async callables:

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any

from agent_adapter.core.context import ToolContext
from agent_adapter.core.tool import ToolSpec
from agent_adapter.core.types import ApprovalDecision
from agent_adapter.exec.defaults import runner_for
from agent_adapter.integrations.base import IntegrationBase, ToolRunner


@dataclass(frozen=True)
class DemoCall:
    """The framework's record of one tool call; `approval` is set by its own approval gate."""

    id: str
    arguments: Mapping[str, Any] = field(default_factory=dict)
    approval: ApprovalDecision | None = None


@dataclass(frozen=True)
class DemoTool:
    """The framework's tool: what the model sees, plus an async call."""

    name: str
    description: str
    parameters: dict[str, Any]
    spec: ToolSpec = field(repr=False)
    runner: ToolRunner = field(repr=False)
    integration: DemoIntegration = field(repr=False)

    async def __call__(self, call: DemoCall) -> Any:
        # bind_context -> runner.run_spec(..., call_id=call.id); the framework has no
        # tool-error channel, so model-visible errors come back as the result.
        return await self.integration._run_bound(call, self.spec, call.arguments, self.runner)


class DemoIntegration(IntegrationBase):
    """Exports registry tools as `DemoTool`s that run through the given runner."""

    name = "demo"  # the entry-point name
    call_id_fields = ("id",)  # DemoCall.id becomes ctx.call_id

    def export_tool(self, spec: ToolSpec, *, executor: ToolRunner | None = None) -> DemoTool:
        """One `DemoTool` for `spec`."""
        runner = runner_for(executor)  # the given runner, else the default Executor
        return DemoTool(spec.name, spec.description, spec.json_schema(), spec, runner, self)

    def bind_context(self, native: Any) -> ToolContext:
        """The base context plus the framework's own approval, so the executor skips its port."""
        ctx = super().bind_context(native)
        ctx.approval = getattr(native, "approval", None)
        return ctx


INTEGRATION = DemoIntegration()

Rules every exported tool follows:

Run through the runner (_run_bound does): runner.run_spec(spec, args, ctx, call_id=…). Never call spec.invoke directly, or hooks, approval, timeouts and the ledger are skipped. Default executor=None to exec.defaults.runner_for(executor), as the built-ins and the demo do.
Use a context from bind_context, so ctx.call_id is the framework's call id.
Export registry.for_target(name) only (IntegrationBase.export_tools does this).
Never catch BaseException: asyncio.CancelledError and LangGraph's GraphBubbleUp must pass through.

2. Report errors the framework's way¶

Framework has…	`ToolArgsInvalid`, `ToolError`, `ApprovalDenied`, `GuardrailBlocked`, `RunLimitExceeded`	Every other exception
a native tool-error channel (LangChain `ToolException`, MCP `is_error`, …)	that channel	propagates
none	return `model_visible_error(exc)` (`MODEL_VISIBLE_ERRORS` lists the classes)	propagates

model_visible_error(exc) is {"error": {"kind": …, "message": …}}; a foreign exception shows only its class name, never its text.

3. Approval¶

Framework	Do
Has its own approval gate	After the gate approves, put the decision in `ctx.approval`; the executor then skips its port
Has none	Return `None` from `bind_approval`; the executor asks its `ApprovalPort`

4. Register the entry point¶

In your package's pyproject.toml:

[project.entry-points."agent_adapter.integrations"]
demo = "demo_integration:INTEGRATION"

The value is a module attribute (an instance) or a class (instantiated with no arguments).

from agent_adapter.integrations.discovery import available_integrations, get_integration

# IntegrationError if the name is unknown; MissingDependency if its extra is missing.
openai = get_integration("openai")
# Protocol-only API: export the tools whose `targets` include this integration.
tools = [openai.export_tool(spec) for spec in registry.for_target(openai.name)]
# {name: integration}; cached, refresh=True reloads.
names = available_integrations()

Discovery never fails because of one plugin. It logs a WARNING and skips an entry that:

Problem	Example
Module or attribute missing	the plugin is half-installed
Import or constructor raises	a bug in the plugin
Not an `Integration`	wrong object registered
Bad attribute types, or `name` ≠ entry-point name	`requires` is a list
Name already taken	two plugins claim `demo` (the first wins)

5. Pass the contract kit¶

tests/integrations/kit.py runs eight checks through your framework's own call path. It is test code, not part of the wheel: a third-party package copies (vendors) tests/integrations/kit.py, tests/integrations/_kit_support.py and tests/fakes/{hooks,approval}.py into its own tests/ package. Write a harness and install the fixture and tests contract_tests returns:

class DemoHarness:
    """Drives the demo framework for the contract kit."""

    name = "demo"
    edits_supported = True  # the executor applies edited args
    approval = "port"  # the executor's approval port decides

    def export(self, spec: ToolSpec, executor: ToolRunner) -> Any:
        return demo().INTEGRATION.export_tool(spec, executor=executor)

    async def call(self, native: Any, args: dict[str, Any], *, call_id: str) -> Any:
        return await native(demo().DemoCall(id=call_id, arguments=args))

    def value(self, raw: Any) -> Any:
        return raw

    def is_error(self, raw: Any) -> bool:
        return isinstance(raw, dict) and set(raw) == {"error"}

    async def approve_flow(
        self, native: Any, args: dict[str, Any], decision: ApprovalDecision
    ) -> Any:
        return await self.call(native, args, call_id="call_demo_approval")

    def describe(self, native: Any) -> tuple[str, str, dict[str, Any]]:
        return native.name, native.description, native.parameters

    def expected_schema(self, spec: ToolSpec) -> dict[str, Any]:
        return spec.json_schema()  # the demo framework takes the neutral schema


globals().update((item.__name__, item) for item in contract_tests(DemoHarness))

Check	Asserts
C1	name, description and the target's schema are exported
C2	the happy path returns what `spec.invoke` returns
C3	invalid args give the framework's error, never a raw `pydantic.ValidationError`
C4	executor policy applies: hooks see before/after; a timeout is `tool-timeout`
C5	the native call id reaches `ctx.call_id` (set `native_call_ids = False` if there is none)
C6	`confirm=True`: deny skips the tool, approve runs it, edits apply when supported
C7	`asyncio.CancelledError` propagates
C8	`get_integration(name)` finds the integration through its entry point

Optional harness members (read with getattr; see the Harness docstring):

Member	Use
`native_call_ids = False`	the framework has no call id (C5)
`cancels_tool = False`	sync-only bridge: C7 checks the awaiter is cancelled, not the tool thread
`approval_port()`, `native_uses_port = True`, `port_asks`	a native gate built on an `ApprovalPort` (C6); `port_asks` pins the number of asks
`export_all(registry, executor)`	the module's `*_tools(registry)`; C1 checks `targets`
`__aenter__` / `__aexit__`	one session (server, client) per test module

contract_tests() returns the harness fixture first: when applying marks over its items, skip that first element (fixture, *tests = contract_tests(...)). By default the kit runs against its FakeRunner; pass contract_tests(MyHarness, runner_factory=Executor) to run through the real executor. A built-in integration's test file is tests/integrations/test_<name>_contract.py, starts with pytest.importorskip("<root>") and sets pytestmark = pytest.mark.contract.

Files¶

File	Role
`agent_adapter/integrations/base.py`	`ToolRunner`, `Integration`, `FlowCompiler`, `IntegrationBase`, `model_visible_error`
`agent_adapter/integrations/discovery.py`	`GROUP`, `available_integrations`, `get_integration`
`tests/integrations/kit.py`, `_kit_support.py`	`Harness`, `contract_tests`, `harness_session`; `FakeRunner`, `Probe`
`tests/fixtures/demo_integration/`	the installable demo plugin (`nox -s plugin`)