插件与扩展

按要添加的行为选择扩展：

扩展点	用途
Harness 中间件插件	包装输入、执行、错误或最终 Run 结果
Capability	添加工具、指令或 Agent 循环行为
`EnvironmentSource`	在首次使用时准备一个 Host 目标
`EnvironmentRunExtension`	为已进入的多个挂载执行初始化和清理
Provider 插件	向 Host 提供 Environment Provider

安装包后显式选择扩展。仅安装不会启用。

选择并激活扩展点

通过各扩展对应的 API 选择：

扩展点	选择方式
中间件	向 `build()` 传入具体插件，或启用配置中的 `plugin_key`/`plugin_id`
Capability	加入定义或 Run 组合，见Capabilities
Run 扩展	向 `create_environment_runtime(extensions=...)` 传入实例
Host source	向 `EnvironmentMount` 加入 Host 来源
Provider 插件	在 Host 配置中选择入口点名称，再选择 Provider 类型

AgentSpec 选择 Capabilities；中间件和 Environment 扩展另行配置。

Harness 中间件

包装整个 Run 时使用中间件；Agent 循环内的行为使用Capability。

可以直接传入 AbstractHarnessPlugin 实例，也可以通过选定的 HarnessPluginFactory 入口点创建插件。

包装执行并发出观察事件

实现异步 wrap_run() 并返回 HarnessRunResult。最多等待一次 call_next(exchange)，也可以直接返回完整结果以短路执行。通过 exchange.with_input(...) 替换语义输入。

Python

from typing import Any

from a13n_harness import AbstractHarnessPlugin, HarnessRunResult
from a13n_harness.events import HarnessExtensionEvent
from a13n_harness.plugins import PluginRunExchange, PluginRunNext


class AuditPlugin(AbstractHarnessPlugin):
    def __init__(self, plugin_id: str) -> None:
        self._plugin_id = plugin_id

    @property
    def plugin_id(self) -> str:
        return self._plugin_id

    async def wrap_run(
        self, exchange: PluginRunExchange, call_next: PluginRunNext[Any]
    ) -> HarnessRunResult[Any]:
        await exchange.context.events.emit(
            HarnessExtensionEvent(kind="diagnostic", payload={"phase": "before"})
        )
        result = await call_next(exchange)
        await exchange.context.events.emit(
            HarnessExtensionEvent(kind="diagnostic", payload={"phase": "after"})
        )
        return result

中间件执行期间，Run 通过有界通道投递扩展事件。调用 continuation 之前、之后以及短路执行期间都可以发出事件。消费者较慢时会施加背压；中间件不拥有或迭代事件流。

破坏性变更迁移： 将 PluginRunResponse 和异步事件生成器改为 async def wrap_run(...)、await call_next(...) 及返回结果。通过 context.events.emit() 发出扩展观察事件。原生事件和 Run 生命周期事件仍由 Harness 管理。需要调整展示时，在 Host 共用的实时与持久化 fold 之前配置 Stream Protocol DisplayFold processor，恢复 fold 时使用相同的 processor。不要在中间件中改写原生事件。

直接组合

嵌入式应用最简单的方式是直接传入对象：

Python

from a13n_harness import HarnessBuilder

plugin = AuditPlugin("audit-primary")
executable = HarnessBuilder().build(
    agent_spec,
    output_type=str,
    model=model,
    plugins=(plugin,),
)

功能 Capability 与当前 Provider 客户端

在定义中选择功能 Capability，通过 Run bindings 提供当前客户端：

Python

from a13n_harness import RunBindings
from a13n_harness.capabilities import WebBinding

result = await executable.run(
    "Read the page",
    bindings=RunBindings.embedded(
        web=WebBinding(client=web_client, policy=web_policy),
    ),
)

WebCapability 放在定义中，不通过插件贡献。Host 管理 provider 客户端及其生命周期；HarnessState 不保存客户端。

离线中间件与 Web 示例将记录插件、由定义负责的 Web 功能、新的类型化绑定和标准 fetch 工具组合起来，无需发起网络请求即可演示。

发布插件工厂

注册一个无需参数的工厂类：

TOML

[project.entry-points."a13n_harness.plugins"]
"acme.audit" = "acme_harness.plugin:AuditPluginFactory"

入口点名称必须与 plugin_key() 一致：

Python

from collections.abc import Mapping

from a13n_harness import AbstractHarnessPlugin
from a13n_harness.plugin_factories import (
    HarnessPluginFactory,
    HarnessPluginFactoryContext,
)
from pydantic import BaseModel, JsonValue


class AuditConfiguration(BaseModel):
    mode: str = "metadata"


class AuditPlugin(AbstractHarnessPlugin):
    def __init__(self, plugin_id: str) -> None:
        self._plugin_id = plugin_id

    @property
    def plugin_id(self) -> str:
        return self._plugin_id


class AuditPluginFactory(HarnessPluginFactory):
    @classmethod
    def plugin_key(cls) -> str:
        return "acme.audit"

    def validate_configuration(
        self,
        configuration: Mapping[str, JsonValue],
    ) -> BaseModel:
        return AuditConfiguration.model_validate(dict(configuration))

    def create_plugin(
        self,
        context: HarnessPluginFactoryContext,
    ) -> AbstractHarnessPlugin:
        # context.configuration holds the validated, normalized configuration.
        return AuditPlugin(context.plugin_id)

在 validate_configuration() 验证配置，再创建新插件。工厂构造不执行 I/O，可变 Run 数据放入 for_run()。

通过配置选择插件

Harness 插件文档定义的是数据结构，不要求使用特定文件格式：

Python

from a13n_harness import HarnessBuilder
from a13n_harness.plugin_configuration import HarnessBuildContext

configuration = {
    "schema_version": "1",
    "plugins": [
        {
            "plugin_id": "audit-primary",
            "plugin_key": "acme.audit",
            "enabled": True,
            "configuration": {"mode": "metadata"},
        }
    ],
}

context = HarnessBuildContext.from_configuration(configuration)
builder = HarnessBuilder(build_context=context)

同一结构可以由 YAML 或 JSON 提供：

YAML

schema_version: "1"
plugins:
  - plugin_id: audit-primary
    plugin_key: acme.audit
    enabled: true
    configuration:
      mode: metadata

Python

context = HarnessBuildContext.from_file("harness-plugins.yaml")
builder = HarnessBuilder(build_context=context)

选择入口点键，不使用 module:object 路径。HarnessBuildContext.extensions 向选定工厂传递命名空间 JSON。

可选的环境变量配置来源

配置插件默认关闭。部署可以显式启用由环境变量指定的配置文档：

Terminal

export A13N_HARNESS_PLUGIN_CONFIG_ENABLED=true
export A13N_HARNESS_PLUGIN_CONFIG_FILE=/etc/a13n/harness-plugins.yaml

Python

builder = HarnessBuilder()

来源优先级为 JSON、指定文件、工作目录中的 harness-plugins.yaml。配置缺失或无效时构建失败，仅加载已启用条目。

运行时插件目录

长期运行的 Host 可以将完整安装的发行包发布到新的不可变目录，并在构造替代 builder 前将该目录加入 sys.path：

Python

import importlib
import sys
from pathlib import Path


def activate_plugin_directory(path: str | Path) -> Path:
    plugin_directory = Path(path).resolve()
    normalized = str(plugin_directory)
    if normalized not in sys.path:
        sys.path.append(normalized)
    importlib.invalidate_caches()
    return plugin_directory

目录必须同时包含可导入的包，以及带有入口点的标准发行包元数据。只有一个独立的 .py 文件是不够的。

按以下步骤操作：

发布到尚未纳入搜索路径的新目录。
校验完整安装。
以原子方式放置该目录。
激活该路径。
构建替代可执行对象。

不要原地修改已经导入的版本，也不要添加两个占用同一插件键的版本。

新 builder 解析已选的安装工厂。已有 executable 保留原插件图，待活跃 Run 结束后再释放。

Host 管理包安装、兼容性与回滚。

插件生命周期

插件有三个独立阶段：

配置插件的工厂选择在构造 builder 时完成，插件创建在每次 build() 时为每个根定义或子定义执行；
绑定 Agent：每个构建的根可执行对象或子可执行对象各执行一次；
绑定 Run 并执行中间件：每个逻辑 Run 都重新执行。

从 wrap_run() 返回一个有效结果。在 finally 清理插件资源，继续传播取消和清理失败。Harness 验证每个成功返回的结果。如果 finally 抛出异常，导致替换结果未能返回，则保留最近一次通过验证的内层结果。

在 for_agent() 返回的 Agent 绑定实例中，通过 get_capabilities() 贡献 Capabilities。

工具分组见ToolProxy 插件贡献来源。

Provider 插件

Provider 插件发布 Environment 定义供 Host 选择。Model、Web、Connector 和 Memory 定义通过代码中的 ProviderCatalog 组合。

声明 ProviderManifest(api_version=1)，并在发布的插件中要求 a13n-harness>=0.10.0,<0.11.0。Host 负责目标准备和销毁；connector_factory 返回构造时无副作用的执行 connector。

普通 Harness 中间件的加载方式与 0.9.3 相同，其独立配置封套仍为 schema_version="1"，不与 Provider manifest 共用版本号。导入 Environment API 的中间件仍需完成相应的源码迁移。

声明 Provider 类型、配置、凭据与执行 connector factory：

Python

from a13n_harness.providers.authentication import Authentication, CredentialMode
from a13n_harness.providers.environment.definition import EnvironmentProviderDefinition

ACME_SANDBOX = EnvironmentProviderDefinition(
    type="acme_sandbox",
    display_name="Acme Sandbox",
    configuration_model=AcmeConnectionConfiguration,
    credential_model=AcmeCredential,
    environment_model=AcmeEnvironmentConfiguration,
    connector_factory=_connector,
    describe_environment=_describe,
    authentication=Authentication(mode=CredentialMode.required),
    setup_url="https://acme.example/dashboard",
    setup_label="Acme dashboard",
)

一个发行包为每个入口点导出一个 ProviderManifest：

Python

from a13n_harness.providers.plugins import ProviderManifest

manifest = ProviderManifest(api_version=1, environment=(ACME_SANDBOX,))

TOML

[project.entry-points."a13n_harness.providers.plugins"]
acme = "acme_providers:manifest"

Host 指定自己信任的入口点，并根据内置定义和选定定义构建一个 Environment 目录：

Python

from a13n_harness.providers.catalog import ProviderCatalog
from a13n_harness.providers.plugins import load_provider_plugins

plugins = load_provider_plugins(("acme",))
environments = ProviderCatalog(
    item for plugin in plugins for item in plugin.manifest.environment
)
definition = environments.require("acme_sandbox")

显式选择入口点名称和 Provider 类型。重复类型使目录构建失败，require() 对不可用类型抛出 ProviderNotSelected。

可运行的插件示例在同一个项目中发布一份清单和一个独立的 Harness 中间件插件。已安装 Provider 插件示例演示直接使用和 Harness UI 加载。

Harness 可选依赖

Harness 自带内置 Provider 定义。使用对应后端时安装 extra：

可选依赖	添加内容	使用方
`docker`	Python 的 Docker SDK	`docker` Environment Provider
`e2b`	异步 E2B SDK	`e2b` Environment Provider
`modal`	Modal SDK	`modal` Environment Provider

Terminal

uv add "a13n-harness[docker,e2b]"

读取 Provider 元数据无需厂商 SDK；缺少 extra 时，Provider 打开失败。

环境输入与来源

将 Host 提供的 EnvironmentSource 传入 run(environment=...)。来源的 ensure_ready() 完成准备并返回 connector；参阅环境中的完整示例。使用 EnvironmentMount 添加权限和路径，或使用显式 runtime 动态修改挂载：

Python

from a13n_harness.environment import (
    FILE_ACTIONS,
    EnvironmentMount,
    EnvironmentPermissionSet,
)
from a13n_harness.environment.advanced import create_environment_runtime

environment_runtime = create_environment_runtime(
    mounts={
        "workspace": EnvironmentMount(
            source=source,
            permission_ceiling=EnvironmentPermissionSet(operations=FILE_ACTIONS),
        ),
    },
    default_mount="workspace",
)

permission_ceiling 接收精确操作集合；provider 权限只能进一步收窄。静态挂载注册不准备目标；首次使用时 Harness 打开并拥有一个独立 execution，结束时关闭它。已被其他挂载拥有的 execution 会被拒绝，且不会关闭原有者的资源。

from a13n_harness.environment import (
    EnvironmentRunCallbacks,
    EnvironmentRunExtensionContext,
)


async def prepare_environment(
    context: EnvironmentRunExtensionContext,
) -> None:
    await context.environment.files.write_text(
        "/workspace/.active-run",
        f"{context.run_id}\n",
        mode="create",
    )


async def clean_environment(
    context: EnvironmentRunExtensionContext,
) -> None:
    await context.environment.files.remove(
        "/workspace/.active-run",
    )


active_run_callbacks = EnvironmentRunCallbacks(
    extension_id="workspace-marker",
    on_enter=prepare_environment,
    on_exit=clean_environment,
)

on_enter 在 Harness 发布静态挂载集合后、首次模型请求前运行。仅进入扩展不会准备目标。初始化需要特定操作族时，显式请求就绪：

Python

from a13n_harness.environment import EnvironmentReadinessRequirement

await context.environment.ensure_ready(
    EnvironmentReadinessRequirement(operations=frozenset({"files"}))
)

on_exit 在 adapter 关闭前运行；成功进入后发生失败或取消时也会运行。用它执行清理。

扩展按注册顺序进入，按逆序退出。

自定义资源作用域

如果初始化和清理共享本地状态，或需要更丰富的资源作用域，请使用自定义异步上下文管理器：

Python

from contextlib import asynccontextmanager

from a13n_harness.environment import EnvironmentRunExtensionContext


class WorkspaceMarkerExtension:
    def __init__(self, extension_id: str) -> None:
        self._extension_id = extension_id

    @property
    def extension_id(self) -> str:
        return self._extension_id

    @asynccontextmanager
    async def bind(self, *, context: EnvironmentRunExtensionContext):
        path = "/workspace/.active-run"
        await context.environment.files.write_text(
            path,
            f"{context.run_id}\n",
            mode="create",
        )
        try:
            yield
        finally:
            await context.environment.files.remove(path)

在显式 runtime 上注册直接扩展对象，同时使用普通的 Environment 输入：

Python

from a13n_harness.environment.advanced import create_environment_runtime


environment_runtime = create_environment_runtime(
    mounts={"workspace": environment},
    default_mount="workspace",
    extensions=(WorkspaceMarkerExtension("workspace-marker"),),
)

通过新的 Run 绑定将 runtime 交给 Harness，由 Harness 完成绑定、激活和关闭：

Python

from a13n_harness import RunBindings

bindings = RunBindings.embedded(environment=environment_runtime)
result = await executable.run("Use the prepared workspace", bindings=bindings)

同一个 extensions= 序列可以同时接受回调适配器和自定义扩展对象：

Python

environment_runtime = create_environment_runtime(
    mounts=mounts,
    default_mount="workspace",
    extensions=(
        active_run_callbacks,
        WorkspaceMarkerExtension("custom-marker"),
    ),
)

Host 在构造 adapter 时提供当前 Provider 状态。扩展在 adapter 进入后进入，在 adapter 关闭前按逆序退出。

显式扩展工厂

发行包可以在以下入口点组中注册无副作用的工厂：

TOML

[project.entry-points."a13n_harness.environment_run_extensions"]
"acme.workspace-marker" = "acme_environment.extension:WorkspaceMarkerFactory"

用 build_environment_run_extension_factory_catalog() 选择工厂键，再创建实例：

Python

from a13n_harness import RunBindings
from a13n_harness.environment import (
    EnvironmentRunExtensionFactoryContext,
    build_environment_run_extension_factory_catalog,
)
from a13n_harness.environment.advanced import create_environment_runtime

catalog = build_environment_run_extension_factory_catalog(
    extension_keys=("acme.workspace-marker",),
)
extension = catalog.create_extension(
    EnvironmentRunExtensionFactoryContext(
        extension_key="acme.workspace-marker",
        extension_id="workspace-marker-primary",
        configuration={"marker_path": "/workspace/.active-run"},
    )
)
environment_runtime = create_environment_runtime(
    mounts=mounts,
    default_mount="workspace",
    extensions=(extension,),
)
bindings = RunBindings.embedded(environment=environment_runtime)
result = await executable.run("Use the prepared workspace", bindings=bindings)

extension_key 选择工厂，extension_id 在 runtime 内必须唯一。直接传入对象不需要入口点发现。

可运行示例

集成包示例演示安装式与直接组合，并提供离线测试。

选择并激活扩展点

Harness 中间件

包装执行并发出观察事件

直接组合

功能 Capability 与当前 Provider 客户端

发布插件工厂

通过配置选择插件

可选的环境变量配置来源

运行时插件目录

插件生命周期

Provider 插件

Harness 可选依赖

环境输入与来源

Host 准备回调

Environment Run 扩展

组合回调

自定义资源作用域

显式扩展工厂

可运行示例

本页目录

选择并激活扩展点复制章节链接

Harness 中间件复制章节链接

包装执行并发出观察事件复制章节链接

直接组合复制章节链接

功能 Capability 与当前 Provider 客户端复制章节链接

发布插件工厂复制章节链接

通过配置选择插件复制章节链接

可选的环境变量配置来源复制章节链接

运行时插件目录复制章节链接

插件生命周期复制章节链接

Provider 插件复制章节链接

Harness 可选依赖复制章节链接

环境输入与来源复制章节链接

Host 准备回调复制章节链接

Environment Run 扩展复制章节链接

组合回调复制章节链接

自定义资源作用域复制章节链接

显式扩展工厂复制章节链接

可运行示例复制章节链接

本页目录

选择并激活扩展点

Harness 中间件

包装执行并发出观察事件

直接组合

功能 Capability 与当前 Provider 客户端

发布插件工厂

通过配置选择插件

可选的环境变量配置来源

运行时插件目录

插件生命周期

Provider 插件

Harness 可选依赖

环境输入与来源

Host 准备回调

Environment Run 扩展

组合回调

自定义资源作用域

显式扩展工厂

可运行示例