nMaroulis/protolink

Source: https://github.com/nMaroulis/protolink

ProtoLink

ProtoLink is a lightweight, A2A-first Python framework for building pluggable agents and multi-agent systems. It began as an A2A-based alternative to chain-centric frameworks such as LangChain: instead of organizing an application around chains of model calls, ProtoLink treats each Agent as a self-contained runtime entity with identity, capabilities, lifecycle, tools, optional reasoning, and direct task-based communication.

A2A is the architectural core, not a bolt-on integration. ProtoLink’s native AgentCard, Task, Message, Part, and Artifact runtime model was originally built on A2A 0.3, then extended for inference, tools, structured agent flows and operational modules without abandoning those protocol primitives. That choice keeps the Agent/Task API simple and the runtime pluggable; a2a=True translates the implemented HTTP surface between ProtoLink’s native model and canonical A2A 1.0 JSON-RPC shapes for standard peers.

The agent is the stable composition surface. Plug in only what that agent needs: an API or local LLM, application knowledge for RAG, built-in, native, or MCP tools, a transport, registry, storage and state, telemetry, authentication, logging, policy, or durable run records. Every module is optional and replaceable through a small public interface.

ProtoLink is deliberately LLM-agnostic and local-first. Provider-native tool calling is used when available; a strict JSON action fallback keeps self-hosted and smaller models on Ollama, llama.cpp, LM Studio, vLLM, or custom backends inside the same infer loop. Changing the model does not require rewriting the agent, its tools, or its communication layer.

The base package has one runtime dependency: Pydantic. HTTP servers, gRPC, hosted model SDKs, MCP, telemetry providers, and other integrations are installed only when you choose them.

Simple by default. Explicit when it matters.

Get started · Concept · API documentation · Examples

Why ProtoLink?

• Pluggable by design - compose an agent from independent modules instead of adopting a mandatory stack.
• A small, stable API - string aliases cover the common path; concrete implementations expose full control when needed.
• Local first, distributed when needed - develop with no network or provider, then move the same task contract to HTTP, SSE JSON-RPC, WebSocket, or gRPC.
• Friendly to smaller models - one-action-at-a-time inference, schema validation, JSON fallback, and deterministic flows reduce reliance on hidden prompt behavior.
• Explicit and inspectable - tool calls, delegation, task state, policy decisions, approvals, runtime events, traces, and reports have typed representations.
• A2A at the core - agents communicate through cards, tasks, messages, parts, and artifacts rather than framework-private graph state.

Focus on the agent’s role and capabilities. ProtoLink handles the infer loop, validated tool execution, delegation, communication, lifecycle, and the operational modules around them.

Start with one agent

Install the HTTP extra:

uv add "protolink[http]"

Create and start a provider-free agent:

from protolink import Agent, AgentCard

planner_agent = Agent(
    card=AgentCard(
        name="planner",
        description="Builds clear execution plans",
        url="http://127.0.0.1:8000",
    ),
    transport="http",
)

planner_agent.start()

No async main(), event-loop setup, model account, or API key is required. start() owns the lifecycle and blocks for a standalone service; use start(background=True) when embedding the agent in another application.

Plug in only what the agent needs

The constructor is the composition surface. This expanded example uses a local Ollama model, registry discovery, SQLite state and run storage, local telemetry, authentication, file logging, dependency-free built-in web search, a native Python tool, and tools from an MCP server:

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="Plans and coordinates work",
        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="Search local notes")
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()

Install the integrations used here with uv add "protolink[http,mcp]"; the Ollama server and example MCP process run separately. web_search() defaults to Brave and reads BRAVE_SEARCH_API_KEY only when invoked. Pass engine="wikipedia" for documented, keyless English Wikipedia search or engine="duckduckgo" for keyless, best-effort DuckDuckGo HTML search. Registering the tool performs no network request. Remove any constructor argument or tool you do not need, or replace it with your own implementation. Different agents in the same mesh can use different models, transports, credentials, storage, policies, and observability backends.

MCPToolAdapter supports local stdio and remote SSE servers. Once registered, MCP tools follow the same schema validation, policy, execution, and telemetry path as native Python tools.

Attach private or application-owned knowledge with the same progressive-control API:

from protolink import create_knowledge

knowledge = create_knowledge(
    "memory",
    name="product_docs",
    description="product manuals and troubleshooting guides",
    sources=["docs/"],
)
planner_agent.add_knowledge(knowledge)

answer = planner_agent.sync.ask("How do I reset a device?")
print(answer.text, answer.citations)

Agent.invoke() lets the model choose the automatically registered search_product_docs tool. Agent.ask() always retrieves first and returns the answer together with normalized hits and citations. Existing Chroma, Pinecone, Qdrant, or custom search systems can be attached without moving their data. See Retrieval-Augmented Generation.

LLM-agnostic, with a strong local focus

For LLM-backed agents, the infer loop is the heart of ProtoLink:

1. The model proposes one next action, including a knowledge search when one is available.
2. ProtoLink parses and validates it.
3. The runtime executes a tool call, agent delegation, or final response.
4. The structured result is added to the task context.
5. The loop repeats until completion or a configured bound is reached.

Providers with reliable native tool calling use it. Local and smaller models can use the JSON fallback, which exposes the same tool_call, agent_call, and final action contract without depending on a provider-specific SDK feature.

agent_call has two delegation modes: tool_call asks another agent to execute one of its tools, while infer asks that agent’s LLM to handle a prompt and initiates the other agent’s infer loop.

from protolink import Agent, AgentCard, create_llm

local_agent = Agent(
    card=AgentCard(
        name="local-assistant",
        description="Runs against a local model server",
        url="runtime://local-assistant",
    ),
    transport="runtime",
    llm=create_llm(
        "ollama",
        base_url="http://127.0.0.1:11434",
        model="qwen3:4b",
    ),
)

Swap "ollama" for another built-in or custom LLM; the agent, tools, tasks, and flows do not change.

A2A primitives, standard wire compatibility

ProtoLink uses A2A’s core AgentCard, Task, Message, Part, and Artifact concepts as first-class Python runtime primitives. Delegation, lifecycle transitions, structured flows, tool results, telemetry, and replay all operate on those explicit objects rather than escaping into a separate orchestration format.

Standard wire compatibility is explicit and additive:

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

# "auto" prefers the full ProtoLink contract and discovers A2A-only peers.
result = await a2a_agent.call_agent(peer_url, task)

# Select the protocol explicitly when the peer protocol is already known.
result = await a2a_agent.call_agent(peer_url, task, protocol="a2a")
result = await a2a_agent.call_agent(peer_url, task, protocol="protolink")

An explicit protocol="a2a" choice bypasses the native-vs-A2A selection step, but still fetches and validates the peer’s standard Agent Card and compatible JSON-RPC interface before sending work.

Agent-originated A2A discovery is always same-origin: an advertised interface must match the Agent Card’s origin. For a split-origin deployment you explicitly trust, use a dedicated AgentClient(..., a2a_allow_cross_origin=True); see A2A compatibility for the operational limits.

With the default a2a=False, HTTP behaves exactly as before: native tasks, status, health, chat, and control endpoints only. With a2a=True, the agent additionally serves the standard Agent Card and SendMessage, GetTask, ListTasks, and CancelTask JSON-RPC operations, and its client can translate outbound calls to A2A-only peers. Outbound ProtoLink infer instructions become A2A user text. Inbound A2A user text remains a normal ProtoLink text part for custom handlers; the default LLM engine recognizes the A2A metadata and treats that text as an inference request. Framework-specific tool-call and flow state should stay on the native protocol.

Compatibility is versioned and testable: the official A2A Technology Compatibility Kit measures the adapter against a pinned protocol surface. The A2A compatibility page records the exact binding, TCK commit, commands, current result, and the remaining upstream harness limitation.

Structured flows

Agents can choose their own next action, but not every workflow should be probabilistic. Pipeline, Parallel, Router, and Graph provide explicit, deterministic topology while keeping every step on the same Task -> Task contract.

from protolink import Pipeline, Task

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

result = review_flow.sync.execute(
    Task.create_infer(prompt="Prepare the release plan"),
)

Flows can contain local agents, registry-resolved remote agents, or other nested flows. Semantic context injection tells each agent what the next step expects without coupling that agent to the overall topology. See structured flows and the runnable examples.

Progressive control

The common path stays small:

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

The alias selects the communication boundary without changing the agent API:

These are qualitative protocol-overhead profiles, not benchmark results; model and tool latency commonly dominate an agent call. See the transport guide for the complete performance, utility, and deployment comparison.

When a boundary needs TLS, resource limits, retries, keepalive settings, or other operational controls, construct the transport and pass it to the same 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 and Registry follow the same rule: pass a string for built-in defaults or a concrete implementation for full control. The façade does not change as deployment requirements grow.

Local telemetry and replay

ProtoLink includes dependency-free local tracing. Attach LocalTraceTelemetry, run a task, and replay the captured spans without sending data to an external service:

from protolink import Agent, AgentCard, LocalTraceTelemetry, create_llm

telemetry = LocalTraceTelemetry(path="traces.jsonl")
agent = Agent(
    AgentCard(name="debug", description="Debug agent", url="runtime://debug"),
    transport="runtime",
    llm=create_llm("mock", default_response="done"),
    telemetry=telemetry,
    verbosity=0,
)

result = agent.sync.invoke("Trace this task")
trace = telemetry.recorder.replay()[-1]

The same runtime contracts power cancellation, budgets, policy decisions, approval previews, run reports, redaction, read-only replay, and normalized report comparison for regression testing. Replay and comparison never re-execute model or tool calls: execute the candidate separately against controlled dependencies, record its report, and then diff it against the baseline. Normalization is limited to known ProtoLink report-envelope fields; application-owned payloads and report metadata remain exact unless you configure an ignore rule or numeric tolerance.

Dashboard developer tool

The protolink dashboard CLI command projects run-store and registry state into a dependency-free local browser UI:

protolink dashboard --store runs.db --registry-url http://127.0.0.1:9010 --open

It reads task snapshots and RunReport records from SQLiteRunStore, loads AgentCard entries from the registry, and provides local views for agent health, HTTP chat, task history, trace summaries, and run replay. The dashboard does not create agents or upload telemetry; it presents the runtime state your agents already emit.

The CLI also includes project scaffolding, environment diagnostics, registry inspection, run replay, and normalized report diffing:

protolink init agent
protolink doctor
protolink run list --store runs.db
protolink run diff baseline_run candidate_run --store runs.db

See the developer tools guide.

Infer-loop benchmark

The ProtoLink repository includes a closed-world regression benchmark for prompt and infer-loop changes. It exercises direct answers, local tools, directed and autonomous agent routing, dependent multi-step work, and grounding traps, then reports strict and recovered functional scores plus end-to-end, model-call, provider, and repeat/cache-sensitive timing.

python -m benchmarks.infer_loop --provider ollama --model gemma4:e4b --suite smoke

The benchmark is source-checkout tooling under benchmarks/, not part of the installable package. See the infer-loop benchmark guide for suite sizes, scoring, Ollama configuration, timing, baseline comparison, filtering, and CI thresholds.

More examples

• Built-in multi-engine web search
• Provider-free runtime mesh
• Normalized run regression diffing
• HTTP agent communication
• Production transport configuration
• Runtime policy and approvals
• Task cancellation
• Structured flows
• All examples

Contributing

Contributions are welcome. See CONTRIBUTING.md and the development guide.

ProtoLink is available under the MIT License.