Build an integration

An integration is one folder under the integrations directory that contributes agent tools (and optionally entities, capability providers, sub-agents, surfaces, and skills) from a config entry. The model is deliberately Home-Assistant-shaped: a manifest.yaml, a config-flow wizard, and an integration object with an async_setup_entry lifecycle.

This page walks through the folder structure, the manifest fields, the integration class and its lifecycle, how the registry discovers and loads integrations, scope and single- vs multi-instance, and a minimal end-to-end example.

For depth on each half, see config-flows.md (the multi-step config wizard, field types, validation, reconfigure) and integration-capabilities.md (entities, capability providers, agents, surfaces, events).

!!! warning “Backend integrations run full-trust, in-process” Discovered integration code is imported and run in-process in the API and worker. It can read the database and the BYOK master key — a malicious integration is a full compromise of every tenant. There is no sandbox for backend integration code. The bundled in-repo dir is always scanned; external drop-in dirs are scanned only when the operator explicitly opts in (see Discovery & loading). Every load is logged at WARNING (integration_loaded_full_trust).

Folder structure

One folder per integration, named exactly after its domain:

integrations/<domain>/
  manifest.yaml          # domain, name, version, config_flow, requirements, …
  __init__.py            # exposes a module-level INTEGRATION instance
  config_flow.py         # ConfigFlow subclass — only if config_flow: true
  tools.py               # builds the pydantic-ai toolset(s)            (as needed)
  providers.py           # capability-provider backends                (as needed)
  entities.py            # entity sync logic                           (as needed)
  translations/<lang>.json     # config-flow labels/errors (server-side i18n)
  skills/<name>/SKILL.md       # optional bundled skill packages             (as needed)

Only manifest.yaml and __init__.py are required. The other modules exist by convention - the loader imports __init__.py as a synthetic package, so intra-package relative imports (from .tools import build_toolset) resolve. Split your code however you like; the conventional split is tools.py / providers.py / entities.py. A skills/<name>/SKILL.md dir (sibling files become skill resources) is parsed at load and projected into the skills table when an entry is set up, scoped to the entry.

!!! note “The folder name must equal the domain” The loader rejects an integration whose manifest.yaml domain does not match its directory name (integration_domain_mismatch).

manifest.yaml fields

The manifest is validated against IntegrationManifest, which sets extra="forbid" — a typo’d key fails loudly at discovery rather than silently dropping config.

iot_class is one of local_in_process, local_polling, local_push, cloud_polling, cloud_push.

!!! warning “required_tier is the single data-governance axis” A typo’d required_tier raises at discovery rather than silently un-gating. An integration carries no tier-gated data unless it says so — declare up for sensitive data (e.g. OpenProject sets required_tier: internal so its work-package data only reaches an internal-tier model).

A minimal manifest:

domain: text_tools
name: Text Tools
version: "0.1.0"
integration_type: tool
config_flow: false
single_instance: true
iot_class: local_in_process
codeowners:
  - "@you"
documentation: https://example.com/docs

The integration class

__init__.py must expose a module-level INTEGRATION that is an instance of a PersonalAgentIntegration subclass. The loader injects manifest and (when config_flow: true) config_flow_cls onto it; you override the lifecycle methods you need.

from __future__ import annotations

from typing import Any

from personal_agent.integrations.integration import PersonalAgentIntegration, SetupContext

from .tools import build_toolset


class TextToolsIntegration(PersonalAgentIntegration):
    async def async_setup_entry(self, ctx: SetupContext) -> list[Any]:
        return [build_toolset()]


INTEGRATION = TextToolsIntegration()

When config_flow: true, point the class at its flow either by class attribute or by exposing a module-level CONFIG_FLOW (the loader picks up either):

class TavilyIntegration(PersonalAgentIntegration):
    config_flow_cls = TavilyConfigFlow

async_setup_entry(ctx) — the lifecycle hook

async_setup_entry is called once per config entry (one configured instance) and returns the pydantic-ai toolsets that entry contributes — typically a single FunctionToolset. Return [] if the integration contributes no direct tools (a pure capability provider or a comms account does this).

Everything the hook needs arrives on the SetupContext:

!!! warning “Never log or serialize ctx.secrets” secrets holds decrypted values (API keys, tokens). It lives in memory only and must never be logged or serialized into Temporal inputs (BYOK Contract #5).

A toolset built from a SetupContext reads non-secret config from ctx.data and secrets from ctx.secrets:

import httpx
from pydantic_ai.toolsets import FunctionToolset
from personal_agent.integrations.integration import SetupContext


def build_toolset(ctx: SetupContext) -> FunctionToolset:
    base = str(ctx.data.get("base_url") or "").rstrip("/")
    auth = httpx.BasicAuth("apikey", ctx.secrets.get("api_key") or "")
    ts = FunctionToolset()

    @ts.tool_plain
    async def my_tool(query: str) -> list[str]:
        """A docstring the model sees — describe what the tool does and returns."""
        async with httpx.AsyncClient(timeout=25.0, auth=auth) as client:
            resp = await client.get(f"{base}/api/search", params={"q": query})
            resp.raise_for_status()
            return [item["title"] for item in resp.json()["items"]]

    return ts

!!! note “Return typed, compact results” Tools should return typed Pydantic models (or an action-result envelope) and cap list sizes so output stays small — never truncate a serialized string. See the openproject tools for the established pattern.

Other lifecycle hooks

All have sensible no-op/empty defaults — override only what you use:

The entity, capability-provider, agent, and surface hooks are covered in integration-capabilities.md.

The config flow

If config_flow: false, the integration is set up without user input (like text_tools). Otherwise add a config_flow.py with a ConfigFlow subclass.

Most flows are a single form with optional connection-test validation, so subclass SimpleConfigFlow and declare only what differs:

from personal_agent.integrations.manifest import FieldDescriptor
from personal_agent.integrations.simple_flow import SimpleConfigFlow

SCHEMA = [
    FieldDescriptor(
        name="api_key",
        label="field.api_key.label",   # an i18n key resolved from translations/<lang>.json
        type="secret",                 # write-only, envelope-encrypted
        required=True,
        description="field.api_key.description",
    ),
]


class MyConfigFlow(SimpleConfigFlow):
    domain = "myservice"
    SCHEMA = SCHEMA
    TITLE_FIELD = "api_key"   # which field becomes the entry title

    async def validate(self, data):
        # Return a localized error KEY ("auth_failed") or None on success.
        # Raising is fine — the exception is logged but NEVER shown to the user.
        return None

Each FieldDescriptor is a JSON-serializable form-field description the frontend renders generically. type is one of text, password, secret, number, bool, select, note, or a typed selector (entity, device, area, duration, date, time, datetime, color). password/secret values are envelope-encrypted (password is shown masked, secret is write-only). note is a read-only display field that can carry an image (e.g. a QR for device linking).

Multi-step flows (e.g. enter a number, then scan a QR, then confirm) subclass ConfigFlow directly and implement async_step_user plus further async_step_<id> methods, returning async_show_form / async_create_entry / async_abort. Accumulated input from prior steps arrives as self._collected. The signal integration is the canonical multi-step example. Full details are in config-flows.md.

Discovery & loading

The registry (IntegrationRegistry) is built once in the API/worker lifespan via IntegrationRegistry.discover(settings.integrations) (which calls loader.discover) and stashed on app.state.integrations (the worker keeps a module-level copy). Discovery:

1. Resolves the dirs to scan: the bundled in-repo dir (bundled_dir, default integrations, resolved relative to the repo root) is always scanned. External drop-in dirs (external_dirs) are scanned only when allow_external is true.
2. For each child folder, parses manifest.yaml against IntegrationManifest, verifies domain == folder name, imports __init__.py as a synthetic package, and resolves the module-level INTEGRATION (must be a PersonalAgentIntegration instance), injecting manifest and config_flow_cls.
3. Checks requirements are importable (a missing one is warned, never installed); if any is missing the integration is kept but marked unavailable (the assembler skips it and a config flow refuses), then logs the load at WARNING.
4. Topologically orders by hard dependencies (an unmet dep drops the integration) and soft after_dependencies.

A bad manifest or a failed import is logged and skipped — boot never crashes. At startup the discovered set is also synced into the integrations table (sync.py): rows are upserted by domain and marked loaded; a domain that disappears is marked not_found (never deleted, so config entries survive a temporary disappearance). Admin-owned columns (enabled, allowed_scopes) are never overwritten by the sync.

The relevant settings (env prefix PERSONAL_AGENT__INTEGRATIONS__):

Single- vs multi-instance, and scope

Instances. single_instance: true (the default) means one config entry per scope/owner. Set single_instance: false to allow several — e.g. two email or two signal accounts. Each entry gets its own async_setup_entry(ctx) call with its own ctx.data / ctx.secrets, so write your build_toolset(ctx) to be per-entry.

Scope. A config entry is owned by a scope: user, org, global, or group. ctx.scope, ctx.owner_sub, and ctx.org_id reflect it. Scope decides who can use the entry’s tools and how its contributions are projected:

• A user entry → user-scoped tools, and contributions (skills / agents / surfaces) owned by that user.
• org / global entries → owner-less contributions visible to the whole scope (org entries keep their org_id).

Which scopes may configure an integration is an admin-governed property (allowed_scopes on the integration row), not a manifest field.

End-to-end example

A minimal service-type integration with a config flow and a single tool. Folder integrations/myservice/:

manifest.yaml:

domain: myservice
name: My Service
version: "0.1.0"
integration_type: service
config_flow: true
single_instance: true
iot_class: cloud_polling
codeowners:
  - "@you"
documentation: https://example.com/myservice

config_flow.py:

from __future__ import annotations

from personal_agent.integrations.manifest import FieldDescriptor
from personal_agent.integrations.simple_flow import SimpleConfigFlow

SCHEMA = [
    FieldDescriptor(
        name="base_url",
        label="field.base_url.label",
        type="text",
        required=True,
        placeholder="field.base_url.placeholder",
    ),
    FieldDescriptor(
        name="api_key",
        label="field.api_key.label",
        type="secret",
        required=True,
    ),
]


class MyServiceConfigFlow(SimpleConfigFlow):
    domain = "myservice"
    SCHEMA = SCHEMA
    TITLE_FIELD = "base_url"

tools.py:

from __future__ import annotations

import httpx
from pydantic import BaseModel
from pydantic_ai.toolsets import FunctionToolset

from personal_agent.integrations.integration import SetupContext


class Hit(BaseModel):
    """One search hit: ``title`` and ``url``."""

    title: str
    url: str


def build_toolset(ctx: SetupContext) -> FunctionToolset:
    base = str(ctx.data.get("base_url") or "").rstrip("/")
    key = ctx.secrets.get("api_key") or ""
    ts = FunctionToolset()

    @ts.tool_plain
    async def myservice_search(query: str) -> list[Hit]:
        """Search My Service. Returns a list of Hit (title, url)."""
        async with httpx.AsyncClient(
            timeout=20.0, headers={"Authorization": f"Bearer {key}"}
        ) as client:
            resp = await client.get(f"{base}/api/search", params={"q": query})
            resp.raise_for_status()
            items = resp.json().get("results") or []
        return [Hit(title=i.get("title") or "", url=i.get("url") or "") for i in items]

    return ts

__init__.py:

from __future__ import annotations

from typing import Any

from personal_agent.integrations.integration import PersonalAgentIntegration, SetupContext

from .config_flow import MyServiceConfigFlow
from .tools import build_toolset


class MyServiceIntegration(PersonalAgentIntegration):
    config_flow_cls = MyServiceConfigFlow

    async def async_setup_entry(self, ctx: SetupContext) -> list[Any]:
        return [build_toolset(ctx)]


INTEGRATION = MyServiceIntegration()

Drop the folder under the bundled integrations dir, add a translations/<lang>.json for the field labels, restart the API and worker, and the integration is discovered, synced into the integrations table, and configurable from the UI.

Where to go next

• config-flows.md — multi-step wizards, field/selector types, field- and connection-level validation, reconfigure.
• integration-capabilities.md — entities (pull/push, RAG, actions, events), capability providers (web / comms / compute), and contributing agents and surfaces.