ProtoLink：构建支持原生代理间通信（A2A）的自主 Python 代理

https://t.co/eYr66IvIns

— # nMaroulis/protolink 源码：https://github.com/nMaroulis/protolink

ProtoLink

Python 版本 (https://www.python.org/downloads/) PyPI 版本 (https://pypi.org/project/protolink/) Ruff (https://github.com/astral-sh/ruff) ty (https://github.com/astral-sh/ty) Ask DeepWiki (https://deepwiki.com/nmaroulis/protolink) 许可证：MIT (https://opensource.org/licenses/MIT) PyPI 下载量 (https://pepy.tech/projects/protolink)

ProtoLink 是一个轻量级的、以 A2A (https://a2a-protocol.org/latest/specification/) 为核心的 Python 框架，用于构建可插拔的代理和多代理系统。它最初作为 LangChain 等以链为中心框架的 A2A 替代方案诞生：不再围绕模型调用链来组织应用，而是将每个 Agent 视为一个自包含的运行时实体，具备身份、能力、生命周期、工具、可选的推理能力，以及基于任务的直接通信。A2A 是架构的核心，而非一个可选的集成模块。

ProtoLink 原生的 AgentCard、Task、Message、Part 和 Artifact 运行时模型最初基于 A2A 0.3 (https://a2a-protocol.org/v0.3.0/specification/) 构建，随后扩展以支持推理、工具、结构化代理流程和运营模块，但从未抛弃这些协议原语。这一选择使得 Agent/Task API 保持简洁，且运行时可插拔；a2a=True 会将实现中的 HTTP 接口在 ProtoLink 原生模型与标准 A2A 1.0 (https://a2a-protocol.org/latest/specification/) JSON-RPC 格式之间进行转换，以便与标准对等体通信。

代理是稳定的组合面。只需插入该代理所需的内容：一个 API 或本地 LLM、内置或原生或 MCP 工具、传输层、注册中心、存储与状态、遥测、认证、日志、策略，或持久化的运行记录。每个模块都是可选的，并且可以通过一个小型公共接口进行替换。

ProtoLink 特意设计为 LLM 无关且本地优先。当可用时，会使用提供商原生的工具调用；严格的 JSON 回退机制确保自托管和较小模型（如 Ollama、llama.cpp、LM Studio 或自定义后端）能够在同一个推理循环中运行。更改模型无需重写代理、其工具或通信层。

基础包只有一个运行时依赖：Pydantic。HTTP 服务器、gRPC、托管模型 SDK、MCP、遥测提供商和其他集成仅在您选择时才会安装。

默认简洁，需要时明确。

开始使用 (https://nmaroulis.github.io/protolink/docs/getting-started/) · 概念 (https://nmaroulis.github.io/protolink/docs/concept/) · API 文档 (https://nmaroulis.github.io/protolink/docs/) · 示例 (https://nmaroulis.github.io/protolink/docs/examples/)

为什么选择 ProtoLink？

• 设计上即可插拔 - 通过独立模块组合代理，而非强制采用一套固定技术栈。
• 小巧稳定的 API - 字符串别名覆盖常见路径；具体实现可在需要时提供完全控制。
• 本地优先，按需分布式 - 无需网络或提供商即可开发，然后将同一任务契约迁移到 HTTP、SSE JSON-RPC、WebSocket 或 gRPC。
• 对小模型友好 - 一次一个动作的推理、模式验证、JSON 回退和确定性流程，减少对隐式提示行为的依赖。
• 明确且可检查 - 工具调用、委托、任务状态、策略决策、审批、运行时事件、追踪和报告都具有类型化的表示。
• 以 A2A 为核心 - 代理之间通过卡片、任务、消息、部分和工件进行通信，而非框架私有的图状态。专注于代理的角色和能力，ProtoLink 负责处理推理循环、验证过的工具执行、委托、通信、生命周期以及围绕它们的运营模块。

从单一代理开始

安装 HTTP 扩展包：

uv add "protolink[http]"

创建并启动一个无需提供商的代理：

from protolink import Agent, AgentCard

planner_agent = Agent(
    card=AgentCard(
        name="planner",
        description="构建清晰的执行计划",
        url="http://127.0.0.1:8000",
    ),
    transport="http",
)

planner_agent.start()

无需 async main()、事件循环设置、模型账户或 API 密钥。start() 拥有生命周期，并会阻塞以运行一个独立服务；若要将代理嵌入到另一个应用中，请使用 start(background=True)。

只插入代理所需的内容

构造函数是组合面。以下扩展示例使用了本地 Ollama 模型、注册中心发现、SQLite 状态和运行存储、本地遥测、认证、文件日志、无依赖的内置网络搜索、原生 Python 工具以及来自 MCP 服务器的工具：

from protolink import (
    Agent,
    AgentCard,
    LocalTraceTelemetry,
    SQLiteRunStore,
    create_llm,
)
from protolink.logging import FileLogger
from protolink.security import APIKeyAuth
from protolink.storage import SQLiteStorage
from protolink.tools import web_search
from protolink.tools.adapters import MCPToolAdapter

planner_agent = Agent(
    card=AgentCard(
        name="planner",
        description="规划并协调工作",
        url="http://127.0.0.1:8000",
    ),
    llm=create_llm(
        "ollama",
        base_url="http://127.0.0.1:11434",
        model="gemma4:e4b",
    ),
    transport="http",
    registry="http",
    registry_url="http://127.0.0.1:9000",
    storage=SQLiteStorage("planner.db", namespace="planner"),
    state=["conversation"],
    run_store=SQLiteRunStore("runs.db"),
    telemetry=LocalTraceTelemetry(path="traces.jsonl"),
    authenticator=APIKeyAuth({"dev-key": []}),
    logger=FileLogger("planner.log"),
)

planner_agent.add_tool(web_search())

@planner_agent.tool(name="search_notes", description="搜索本地笔记")
async def search_notes(query: str) -> str:
    return f"Results for {query}"

mcp_adapter = MCPToolAdapter(
    transport="stdio",
    command="python",
    args=["mcp_server.py"],
)

for tool in mcp_adapter.get_tools():
    planner_agent.add_tool(tool)

planner_agent.start()

使用 uv add "protolink[http,mcp]" 安装此处使用的集成包；Ollama 服务器和示例 MCP 进程需单独运行。web_search() 默认使用 Brave，仅在调用时读取 BRAVE_SEARCH_API_KEY。传递 engine="wikipedia" 可使用无密钥的英文维基百科搜索，或 engine="duckduckgo" 使用无密钥的 DuckDuckGo HTML 搜索（尽力而为）。注册工具不会执行任何网络请求。移除任何不需要的构造函数参数或工具，或替换为自定义实现。

同一网格中的不同代理可以使用不同的模型、传输层、凭证、存储、策略和可观测性后端。MCPToolAdapter 支持本地 stdio 和远程 SSE 服务器。注册后，MCP 工具遵循与原生 Python 工具相同的模式验证、策略、执行和遥测路径。

LLM 无关，同时重点关注本地化

对于基于 LLM 的代理，推理循环是 ProtoLink 的核心：

1. 模型提出下一个动作。
2. ProtoLink 解析并验证该动作。
3. 运行时执行工具调用、代理委托或最终响应。
4. 将结构化结果添加到任务上下文中。
5. 循环重复，直到完成或达到配置的边界。

具有可靠原生工具调用的提供商将使用其原生方式。本地和较小模型可以使用 JSON 回退，该回退暴露了相同的 tool_call、agent_call 和 final 动作契约，而无需依赖提供商特定的 SDK 功能。agent_call 有两种委托模式：tool_call 要求另一个代理执行其某个工具；infer 则要求该代理的 LLM 处理一个提示词，并启动该代理的推理循环。

from protolink import Agent, AgentCard, create_llm

local_agent = Agent(
    card=AgentCard(
        name="local-assistant",
        description="针对本地模型服务器运行",
        url="runtime://local-assistant",
    ),
    transport="runtime",
    llm=create_llm(
        "ollama",
        base_url="http://127.0.0.1:11434",
        model="qwen3:4b",
    ),
)

将 "ollama" 替换为另一个内置或自定义的 LLM，代理、工具、任务和流程均无需更改。

A2A 原语，标准线路兼容性

ProtoLink 将 A2A 的核心概念 AgentCard、Task、Message、Part 和 Artifact 作为一等 Python 运行时原语使用。委托、生命周期转换、结构化流程、工具结果、遥测和回放都基于这些显式对象运行，而非逃逸到单独编排格式中。

标准线路兼容性是显式且增量的：

a2a_agent = Agent(card=card, transport="http", a2a=True)

# "auto" 优先使用完整的 ProtoLink 契约，并发现仅支持 A2A 的对等体。
result = await a2a_agent.call_agent(peer_url, task)

# 当对等体协议已知时，显式选择协议。
result = await a2a_agent.call_agent(peer_url, task, protocol="a2a")
result = await a2a_agent.call_agent(peer_url, task, protocol="protolink")

显式选择 protocol="a2a" 会跳过原生与 A2A 的选择步骤，但在发送工作之前仍会获取并验证对等体的标准 Agent Card 和兼容的 JSON-RPC 接口。代理发起的 A2A 发现始终为同源：一个广告的接口必须与 Agent Card 的源匹配。对于您明确信任的跨源部署，请使用专门的 AgentClient(..., a2a_allow_cross_origin=True)；有关操作限制，请参阅 A2A 兼容性 (https://nmaroulis.github.io/protolink/docs/a2a/)。

使用默认的 a2a=False 时，HTTP 的行为与以前完全相同：仅提供原生任务、状态、健康、聊天和控制端点。使用 a2a=True 时，代理还会额外提供标准的 Agent Card 以及 SendMessage、GetTask、ListTasks 和 CancelTask 的 JSON-RPC 操作，并且其客户端可以将出站调用转换为仅支持 A2A 的对等体。

出站的 ProtoLink infer 指令会转换为 A2A 用户文本。入站的 A2A 用户文本对于自定义处理器而言仍然是普通的 ProtoLink 文本部分；默认的 LLM 引擎会识别 A2A 元数据，并将该文本视为推理请求。框架特定的工具调用和流程状态应保留在原生协议上。

兼容性是有版本且可测试的：官方的 A2A 技术兼容性工具包 (https://github.com/a2aproject/a2a-tck) 根据固定的协议表面测量适配器。A2A 兼容性页面 (https://nmaroulis.github.io/protolink/docs/a2a/) 记录了确切的绑定、TCK 提交、命令、当前结果以及剩余的上游框架限制。

结构化流程

代理可以选择自己的下一个动作，但并非所有工作流都应该是概率性的。Pipeline、Parallel、Router 和 Graph 提供了显式的确定性拓扑，同时使每个步骤都基于相同的 Task -> Task 契约。

from protolink import Pipeline, Task

review_flow = Pipeline(
    steps=[researcher_agent, reviewer_agent, planner_agent],
)

result = review_flow.sync.execute(
    Task.create_infer(prompt="准备发布计划"),
)

流程可以包含本地代理、注册中心解析的远程代理或其他嵌套流程。语义上下文注入告诉每个代理下一步期望什么，而无需将该代理耦合到整体拓扑中。请参阅结构化流程 (https://nmaroulis.github.io/protolink/docs/flows/) 和可运行示例 (https://github.com/nMaroulis/protolink/tree/main/examples/structured_flows)。

渐进式控制

常见路径保持简洁：

agent = Agent(card=card, transport="http")

别名无需更改代理 API 即可选择通信边界：

这些是定性的协议开销描述，而非基准测试结果；模型和工具延迟通常在代理调用中占主导地位。请参阅传输指南 (https://nmaroulis.github.io/protolink/docs/transport/) 以获取完整的性能、实用性和部署比较。

当边界需要 TLS、资源限制、重试、保活设置或其他操作控制时，可以构造传输层并将其传递给相同的 API：

from protolink import RetryPolicy, TLSConfig, TransportConfig, TransportLimits
from protolink.transport import HTTPTransport

transport = HTTPTransport(
    url=card.url,
    tls=TLSConfig(
        certfile="certs/agent.pem",
        keyfile="certs/agent-key.pem",
        cafile="certs/ca.pem",
    ),
    config=TransportConfig(
        limits=TransportLimits(max_concurrent_requests=200),
        retry=RetryPolicy(max_attempts=3),
    ),
)

agent = Agent(card=card, transport=transport)

AgentClient 和 Registry 遵循相同的规则：传递字符串以使用内置默认值，或传递具体实现以获得完全控制。随着部署需求增长，外观模式不会发生变化。

本地遥测与回放

ProtoLink 包含无依赖的本地追踪。附加 LocalTraceTelemetry，运行一个任务，然后回放捕获的跨度，无需将数据发送到外部服务：

from protolink import Agent, AgentCard, LocalTraceTelemetry, create_llm

telemetry = LocalTraceTelemetry(path="traces.jsonl")

agent = Agent(
    AgentCard(name="debug", description="调试代理", url="runtime://debug"),
    transport="runtime",
    llm=create_llm("mock", default_response="done"),
    telemetry=telemetry,
    verbosity=0,
)

result = agent.sync.invoke("追踪此任务")

trace = telemetry.recorder.replay()[-1]

相同的运行时契约也支持取消、预算、策略决策、审批预览、运行报告、编辑、只读回放以及用于回归测试的归一化报告比较。回放和比较从不重新执行模型或工具调用：单独针对受控依赖执行候选任务，记录其报告，然后与基线进行差异对比。归一化仅限于已知的 ProtoLink 报告信封字段；应用程序拥有的负载和报告元数据保持精确，除非您