Show HN: ProtoLink – A Python runtime for distributed A2A agents

📌 The framework is currently in

betaand is subject to change.

ProtoLink is a lightweight Python framework that allows you to build autonomous, LLM-powered agents that communicate directly, manage context, and integrate tools seamlessly. Build distributed multi-agent systems with minimal boilerplate and production-oriented architecture.

Each ProtoLink agent is a self-contained runtime that can embed an LLM, manage execution context, expose and consume tools (native or via MCP), and coordinate with other agents over a unified transport layer.

ProtoLink implements and extends Google’s Agent-to-Agent (A2A) specification for agent identity, capability declaration, and discovery, while going beyond A2A by enabling LLM & tool integration.

ProtoLink is an

A2A-native agent runtimefor building distributed, typed, observable agent systems. LangChain composes model calls; ProtoLink runs agents.

Find more in the Protolink whitepaper.

The framework emphasizes minimal boilerplate, explicit control, and production-readiness, making it suitable for both research and real-world systems.

Tool calling, agent delegation, LLM invocation, and task execution logic are provided by Protolink. You describe the agent, its tools, and its capabilities; ProtoLink handles the infer loop, native tool calling where available, JSON fallback for smaller models, action validation, and deterministic execution.

Protolink uses dynamic semantic context injection, automatically enriching agent prompts with downstream agent capabilities, tools, and communication contracts so agents remain fully decoupled while adapting their behavior at runtime without hardcoded integrations.

Focus on your agent logic- ProtoLink handles communication, authentication, LLM integration, and tool management for you.

Follow the API documentation here 📚 documentation.

The following articles published on * Level Up Coding* on

give a hands-on guide and overview of Protolink:

Medium- 📝 Building My Own Local “Claude Code”: What I Learned Demystifying Agentic Coding under the Hood - 📝 Your First Autonomous Agent Mesh - Easier Than You Think - 📝 Build Easily Your Own “Claude Code” with Three Agents: Brain, Hands, and Coordinator

Update 0.5.0 introduces Structured Flows, a new feature that allows you to define structured workflows for your agents. Building on A2A transport layer, flows allow you to define complex workflows for your agents that can be executed in a structured & deterministic manner. Technically, each flow is a deterministic Flow.execute(Task) -> Task

state machine over A2A Task

, Message

, Artifact

, and AgentCard

primitives, keeping orchestration protocol-native instead of creating a separate graph runtime. Again the Semantic Context Injection is handled by Protolink, meaning that the agents will automatically have the necessary context about the flow they are executing and what to pass down to next agents in the pipeline. See more here 🔀 flows.

In Protolink the agent is the central component that handles all the logic and incorporates the LLM, tools, transport layer through AgentClient and AgentServer, the Storage and OpenTelemetry for logging.

Each of these components is a separate module that can be used independently or in combination with other modules. Each component is pluggable to the agent and can be replaced with your own implementation.

Here's a simple example on how easy it is to create an agent, define the neccessary modules, provided by Protolink and plug 🔌 them to the agent. The agent is then ready to discover other agents and send & receive Tasks.

from protolink.agents import Agent

from protolink.discovery import Registry
registry = Registry(url="http://127.0.0.1:9000", transport="http")
registry.start(background=True)

from protolink.llms.api import OpenAILLM
llm = OpenAILLM(model="gpt-5.2") # for local model use protolink.llms.server.OllamaLLM and more...

from protolink.storage import SQLiteStorage
storage = SQLiteStorage(db_path="agent.db")

from protolink.telemetry import LangfuseTelemetry
telemetry = LangfuseTelemetry()

agent_card = {
    "url": "http://127.0.0.1:8020",
    "name": "example_agent",
    "description": "A dummy agent",
}

agent = Agent(card=agent_card, transport="http", llm=llm, registry=registry, storage=storage, telemetry=telemetry)

@agent.tool(name="add", description="Add two numbers")
async def add_numbers(a: int, b: int):
    return a + b

agent.start()

The agent is now fully initialized and prepared to discover & be discovered by peers, send & receive tasks across your system.

Running tasks can be canceled by task ID across local and remote transports, with final canceled state propagated through RunContext

and streamed runtime events.

Note: agent.start()

automatically adapts to the current environment under the hood, working seamlessly in standard Python scripts, async applications, and Jupyter notebooks with both blocking and background execution modes. When background=True

, it runs the agent non-blocking using an event loop task or a dedicated background thread depending on the runtime context. See how Protolink handles agent execution lifecycle.

A2A Protocol Implementation: Fully compatible with** Google's A2A specificationExtended Capabilities**:** Unified Client/Server Agent Model**: Single agent instance handles both client and server responsibilities, reducing complexity.** Transport Layer Flexibility**: Swap betweenHTTP,WebSocket,gRPCorin-memorytransports with minimal code changes.Simplified Agent Creation and Registration: Create and register** autonomous AI agentswith just a few lines of code. LLM-ReadyArchitecture: Native support for integrating LLMs to agents (APIs & local) directly as agent modules, allowing agents to expose LLM calls, reasoning functions, and chain-of-thought utilities with zero friction.Tooling: Native supportfor integrating tools to agents (APIs & local) directly as agent modules. Native Adapter for MCP tooling**.** Runtime Transport Layer**: In-process agent communication using a shared memory space. Agents can easily communicate with each other within the same process, making it easier to build and test agent systems.Enhanced Security: Native** TLS/mTLS**,** OAuth 2.0**, bearer tokens, Basic auth, and** API key support**.** Comprehensive Logging**: Built-in support for** console**(colored),** file-basedlogging (including structured JSON**output), and a no-op quiet logger.- Built-in support for streaming and async operations.

Planned Integrations:** Advanced Orchestration Patterns**- Multi-step workflows, supervisory agents, role routing, and hierarchical control systems.

Runtime Control Plane: Typed client/server requests for cancellation, state inspection/reset/compaction, and LLM history compaction without exposing maintenance operations to the model prompt.Observable & Replayable Runs:RunContext

,RunEvent

,RunReport

, budgets, policy decisions, approvals, and redaction for production-facing UIs and tests.Developer Tools:protolink doctor

, registry inspection, run replay, and a local dashboard with agent ping, HTTP chat, registry health, and a disabled Studio preview for future topology work.

Protolink exposes a dashboard UI through the protolink dashboard

CLI command, projecting run-store and registry state into a local browser view.

Run the dashboard against an existing run store and a live registry (both optional):

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

The command serves a dependency-free local HTML dashboard over Protolink's public devtool collectors. It reads persisted task snapshots and RunReport

records from SQLiteRunStore

, pulls AgentCard

entries from the registry, and exposes live dashboard actions through local endpoints for snapshot refresh, run replay, HTTP agent /status

probes, and standard agent chat. The page does not create agents; it projects the runtime state your agents already emit through AgentCard

, RunContext

, RunEvent

, and RunReport

.

ProtoLink implements Google’s A2A protocol at the wire level, while providing a higher-level agent runtime that unifies client, server, transport, tools, and LLMs into a single composable abstraction the Agent.

Concept	Google A2A	ProtoLink
Agent	Protocol-level concept	Runtime object
Transport	External server concern	Agent-owned
Client	Separate	Built-in
LLM	Out of scope	First-class
Tools	Out of scope	Native + MCP
UX	Enterprise infra	Developer-first

Protolink takes a centralized agent approach compared to Google's A2A protocol, which separates client and server concerns. Here's how it differs:

Feature	Google's A2A	Protolink
Architecture
Decoupled client/server	Unified agent with built-in client/server
Transport
Factory-based with provider pattern	Direct interface implementation
Deployment
Requires managing separate services	Single process by default, scales to distributed
Complexity
Higher (needs orchestration)	Lower (simpler to reason about)
Flexibility
Runtime configuration via providers	Code-based implementation
Use Case
Large-scale, distributed systems	Both simple and complex agent systems

Simplified Development: Manage a single agent runtime without separate client/server codebases.** Reduced Boilerplate**: Common functionality is handled by the baseAgentclass, letting you focus on agent logic.Flexible Deployment: Start with a single process, scale to distributed when needed** Unified State Management**: Shared context between client and server operations** Maintainability**:- Direct code paths for easier debugging

• Clear control flow with fewer abstraction layers
• Type-safe interfaces for better IDE support

Extensibility:- Easily add new transport implementations

• Simple interface-based design
• No complex configuration needed for common use cases

Real Multi-Agent Systems: Build** autonomous agentswith embedded LLMs, tools, and memory that communicate directly. Simple API**: Built from the ground-up for** minimal boilerplate**, letting you focus on agent logic rather than infrastructure.** Developer Friendly**: Clean abstractions and direct code paths make debugging and maintenance a breeze.** Production Oriented**: Designed for** performance, reliability, and scalabilityin real-world deployments. Extensible & Interoperable**: Add new agents, transports, or protocols easily; compatible with** A2Aand MCPstandards. Community Focused**: Designed for the open-source community with clear contribution guidelines.

This will install the base package without any optional dependencies.

uv add protolink

pip install protolink

Protolink supports optional features through extras. Install them using square brackets: Note: uv add

can be replace with pip install

if preferred.

uv add "protolink[all]"

uv add "protolink[http]"

uv add "protolink[grpc]"

uv add "protolink[llms]"

uv add "protolink[dev]"

To install from source and all optional dependencies:

git clone https://github.com/nmaroulis/protolink.git
cd protolink
uv pip install -e ".[dev]"

👉 The example found in the jupyter notebooks here: Hello World Example

from protolink.agents import Agent
from protolink.models import AgentCard
from protolink.tools.adapters import MCPToolAdapter
from protolink.llms.api import OpenAILLM
from protolink.discovery import Registry

registry = Registry(url="http://127.0.0.1:9000", transport="http")
registry.start(background=True)

agent_card = AgentCard(
    name="example_agent",
    description="A dummy agent",
    url="http://127.0.0.1:8020",
)

llm = OpenAILLM(model="gpt-5.2")

agent = Agent(agent_card, transport="http", llm=llm, registry=registry)

@agent.tool(name="add", description="Add two numbers")
async def add_numbers(a: int, b: int):
    return a + b

mcp_adapter = MCPToolAdapter(transport="sse", url="https://api.example.com/mcp/sse")
mcp_tools = mcp_adapter.get_tools()
for mcp_tool in mcp_tools:
    agent.add_tool(mcp_tool)

agent.start()

When using an HTTP-compatible transport (http

, sse

, json-rpc

, or sse-json-rpc

), the Agent and Registry expose lightweight browser pages: registry status at /status

, agent status at /status

, and agent chat at /chat

for LLM-backed agents. WebSocket, gRPC, and runtime transports keep the same logical endpoints for clients, but they do not serve browser HTML directly.

| |

ProtoLink's core philosophy is a simple API with progressive control: the shortest path should be enough for prototyping, while production features remain available through explicit, composable objects instead of adding every advanced option to Agent

.

For a fast prototype, pass a transport name. ProtoLink builds the transport with safe defaults:

from protolink import Agent, AgentCard

card = AgentCard(
    name="assistant",
    description="General-purpose assistant",
    url="http://127.0.0.1:8000",
)

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

For production, configure the transport directly and pass the completed object to the same Agent API:

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

card = AgentCard(
    name="assistant",
    description="Production assistant",
    url="https://agent.internal:8443",
)
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)

This keeps Agent concerns such as reasoning, tools, state, and policy separate from transport concerns such as TLS, payload limits, retries, keepalive, and connection management. The same rule applies to AgentClient

and Registry

: pass a string for defaults or a configured transport object for advanced behavior. Every service boundary can therefore use its own certificates and capacity policy without expanding or coupling the high-level constructors.

Follow the API documentation here: Documentation

For Agent-to-Agent & Agent-to-Registry communication:

http

·HTTPTransport: Uses HTTP/HTTPS for synchronous requests. Two ASGI implementations are available.- Lightweight: starlette

,httpx

&uvicorn

• Advanced | Schema Validation: fastapi

,pydantic

&uvicorn

• Lightweight: sse

,json-rpc

,sse-json-rpc

·SSEJSONRPCTransport: Uses normal HTTP routes plus Server-Sent Events for streamed task updates.websocket

·WebSocketTransport: Uses WebSocket for streaming requests. [websockets

]grpc

·GRPCTransport: Uses gRPC unary calls and unary-stream task events over compact JSON envelopes. [grpcio

]runtime

·RuntimeTransport: Simplein-process, in-memory transport.

All network transports share one native TLS API. Use https://

, wss://

, or grpcs://

and configure TLSConfig

on the transport:

from protolink import Agent, AgentCard, TLSConfig
from protolink.transport import HTTPTransport

tls = TLSConfig(
    certfile="certs/agent.pem",
    keyfile="certs/agent-key.pem",
    cafile="certs/ca.pem",
)

card = AgentCard(
    name="secure-agent",
    description="Agent served over HTTPS",
    url="https://agent.internal:8443",
)
transport = HTTPTransport(
    url=card.url,
    tls=tls,
)
agent = Agent(card=card, transport=transport)

TLSConfig

encrypts the connection and verifies certificates; Authenticator

implementations handle application credentials and authorization. They are independent and can be combined. See the TLS and mutual TLS guide and examples/tls_agent.py.

All transports also share one production configuration surface:

from protolink import Agent, AgentCard, RetryPolicy, TransportConfig, TransportLimits
from protolink.transport import HTTPTransport

agent_card = AgentCard(
    name="production-agent",
    description="Production task worker",
    url="http://127.0.0.1:8000",
)
transport_config = TransportConfig(
    limits=TransportLimits(max_concurrent_requests=200, max_concurrent_streams=50),
    retry=RetryPolicy(max_attempts=3),
)

transport = HTTPTransport(url=agent_card.url, config=transport_config)
agent = Agent(card=agent_card, transport=transport)

Retries remain off by default and run only for explicitly idempotent request specifications. Correlation IDs, server-side idempotency replay, bounded payloads/concurrency, loop-safe pooled-resource shutdown, WebSocket keepalive, typed transport errors, local metric snapshots, and /healthz

and /readyz

probes use the same contract across HTTP, SSE JSON-RPC, WebSocket, gRPC, and RuntimeTransport. The grpc

extra also installs standard gRPC health checking and reflection. See the production transport guide and examples/transport_production.py.

These controls exist so changing protocols does not change the Agent's production safety model. Limits prevent one payload or traffic burst from exhausting the process; retries recover explicitly safe operations from temporary connection failures; idempotency prevents those retries from executing the same operation twice; and metrics and health probes make the behavior visible. The defaults are suitable for getting started and do not enable retries automatically.

The application-facing transport types are exported directly from protolink

:

API	Purpose
TransportConfig
One immutable configuration for limits, retries, keepalive, shutdown, idempotency replay, and metrics.
TransportLimits
Request, response, stream-event, request-concurrency, and stream-concurrency bounds.
RetryPolicy
Explicit attempt count, exponential backoff, jitter, and retryable method set.
TransportMetricsSnapshot
Dependency-free per-instance counters, byte totals, active work, retries, and cumulative latency.
TransportError and typed subclasses
Stable connection, timeout, protocol, remote, and payload-limit failures with request metadata.

Transport authors can additionally import Transport

, TransportCapabilities

, and TransportRequestContext

from protolink.transport

. The shared API reference lists every field, default, base-class hook, health payload, and protocol mapping.

Protolink separates LLMs into three types: api

, local

, and server

. The following are the Protolink wrappers for each type. If you want to use another model, you can use it directly without going through Protolink’s LLM

class.

[ API ] [ Server ] [ Local ]

API, calls the API, requires an API key:OpenAILLM: UsesOpenAI API for sync & async requests.AnthropicLLM: UsesAnthropic API for sync & async requests.GeminiLLM: UsesGemini API for sync & async requests.GrokLLM: UsesGrok API for sync & async requests.DeepSeekLLM: UsesDeepSeek API for sync & async requests.

Local, runs the model in runtime:LlamaCPPLocalLLM: Uses** local runtime llama-cpp-python**for sync & async requests.

Server, connects to an LLM Server, deployed locally or remotely:OllamaLLM: UsesOllama for sync & async requests.LlamaCPPServerLLM: Connects tollama-server for sync & async requests.

Native Tool: Uses native tools.MCPToolAdapter: Connects to MCP Server and registers MCP tools as native tools.

The MCPToolAdapter

connects to Model Context Protocol (MCP) servers and exposes their tools as Protolink-native callables, wrapping remote MCP tools as native Tool

instances. This enables seamless integration with the growing MCP ecosystem.

Supported transports:

stdio

• Local subprocess communication (for MCP servers as scripts/executables)sse

• Remote HTTP Server-Sent Events (for MCP servers as web services)

from protolink.tools.adapters import MCPToolAdapter

adapter = MCPToolAdapter(
    transport="stdio",
    command="python",
    args=["my_mcp_server.py"]
)

adapter = MCPToolAdapter(
    transport="sse",
    url="https://api.example.com/mcp/sse"
)

for tool in adapter.get_tools():
    agent.add_tool(tool)

add_tool = adapter.get_callable("add")
result = add_tool(a=5, b=7)

Protolink treats agentic systems as distributed programs, not probabilistic workflows. Every interaction between models, tools, and agents becomes an explicit action that the runtime can validate, execute, observe, and replay. Runs can also carry a typed RunContext

and emit versioned RunEvent

s, giving applications stable session, trace, budget, permission, and progress metadata without custom glue. RunAction

, capability policy, and typed approval checkpoints gate side effects while keeping approval previews and user experience application-owned.

At the heart of Protolink is the infer loop:

• The LLM proposes exactly one next action.
• Protolink validates that action.
• The runtime executes the tool call, agent delegation, or final response.
• The result is fed back into the conversation.
• The loop repeats until the agent returns a final answer.

This is where Protolink removes the boilerplate most agent frameworks push onto you: provider-specific tool prompts, JSON parsing, schema validation, retries, routing, delegation, and error recovery.

When a provider supports native tool calling, Protolink uses it. When a local or smaller model works better with simple JSON instructions, Protolink uses that instead. The user-facing contract stays the same.

Protolink removes that complexity by standardizing all interactions through a small set of explicit primitives:

Task- a shared unit of work** Message**- communication within a task** Part**- an atomic, machine-interpretable action or result

Agents never infer behavior implicitly. Instead, they declare intent explicitly using structured Parts such as:

tool_call- execute a local tool** agent_call**- delegate work to another agent** infer**- invoke LLM reasoning** text**- return user-facing output

and more...

From there, the runtime handles everything deterministically.

You do not need to:

• Write provider-specific tool-calling prompts
• Parse raw LLM text or JSON
• Convert native tool calls into runtime actions
• Write routing ordelegation logic - Rebuild retry, validation, and self-correction loops

The runtime automatically:

• Selects the right prompt/tool-calling mode for the model
• Injects tool schemas and discovered agent capabilities
• Validates every LLM action before execution
• Executes tools and agent calls deterministically
• Emits structured events for tracing and debugging

Tool calls, agent calls, and LLM invocations only happen when explicitly declared. All results are returned as structured Parts - no hidden side effects, no magic. From the user’s perspective:

task = Task.create(
    Message.user("What's the weather in Geneva?")
)

That’s it.

This is not a black-box agent framework.

• No hidden reasoning
• No implicit planning
• No speculative execution

Every action is explicit, inspectable, and replayable.

If a tool runs, you see a tool_call

. If an agent is contacted, you see an agent_call

. If an LLM is invoked, you see an infer

.

This makes the system predictable, debuggable, composable, and production-ready.

The runtime is fully LLM-agnostic. Any model, API-based, self-hosted, or local, can be swapped in without changing behavior or results. OpenAI, Anthropic, local servers, or custom backends all operate through the same unified execution model.

The orchestration stays the same. The contracts stay the same. Only the model changes.

This lets you evolve providers, costs, latency, or deployment strategy without rewriting your agents.

This project uses a structured, Agent-to-Agent (A2A) style communication model. Understanding how Tasks, Messages, Artifacts, and Parts interact is key to using the agent effectively.

A Task represents a unit of work or a conversation thread between agents.

• It contains Messages andArtifacts. - Tracks lifecycle through Task.state

(submitted

,working

,completed

,failed

, etc.). - Records successful state transitions in metadata["state_history"]

. - Tasks are sent between agents; each agent executes what is explicitly defined in the task.

A Message is a communicative unit in a task.

• Can be sent by a user or an agent.

• Contains Parts representing atomic content. - Example roles: "user"

• input from a human or another agent"agent"

• output from an agent

An Artifact is a container for outputs generated by the agent.

• Stores Parts that result from executing a tool (tool_call) or an LLM inference (** infer**). - Can include tool results, reasoning traces, or structured outputs.
• Artifacts allow agents to append results without modifying the original message.

A Part is the atomic content of a Message or Artifact.

• Defines what to do orwhat was produced. - Example Part types ( PartType

):"text"

: plain text content"json"

: structured data"tool_call"

: request to execute a registered tool"tool_output"

: result outputfrom executing a tool"infer"

: input to invoke the agent's LLM"status"

,"error"

,"image"

,"audio"

, etc.

Task Creation

A user or agent creates aTask

with aMessage

containing one or moreParts

. - Task Execution- The receiving agent inspects the last message or artifact in the task. - Executes each Part

sequentially:tool_call

→ executes a registered tool → producestool_output

Part in an Artifact.infer

→ invokes the agent's LLM → producesinfer_output

Part in an Artifact.

• The receiving agent inspects the

Appending Outputs- Results are appended to the Task as new Artifacts. - Lifecycle state transitions are applied to Task.state

and recorded inmetadata["state_history"]

.

• Results are appended to the Task as new

Sequential Processing- Tasks are processed sequentially at the message/artifact level.

• Parallel execution is possible within the parts of a single message/artifact, but not across multiple messages/artifacts in the same task.

from protolink.models import Message, Part, Task

task = Task.create(Message.user("What's the weather in Athens?"))

tool_part = Part.tool_call(tool_name="weather_api", args={"city": "Athens"})
task.add_message(Message.agent(parts=[tool_part]))

result_task = await agent.execute_task(task)

for artifact in result_task.artifacts:
    for part in artifact.parts:
        if part.type == "tool_output":
            print("Tool Output:", part.content)

infer_part = Part.infer(prompt="Summarize today's weather in Athens")
task.add_message(Message.agent(parts=[infer_part]))
result_task = await agent.execute_task(task)

Key Notes:

• Each Task

maintains the full history of messages and artifacts. - Agents execute the last message or artifact to determine the next action. - Parts inside a message or artifact can be executed in parallel if needed.

• Agents do not guess intent; they execute exactly what is defined in the Parts.

This structured approach ensures predictable, deterministic agent behavior while still supporting multi-step interactions and LLM/tool execution.

While frameworks like LangChain or LangGraph rely on complex implicit state machines or LLM-driven routing, Protolink's Structured Flows provide explicit, deterministic execution paths (Pipeline

, Parallel

, Router

, Graph

) without the heavy overhead.

Flows allow you to define highly composable, rigid, and predictable multi-step workflows. Pipeline

, Parallel

, and Graph

define deterministic topology, while Router

branches on structured Part.route(...)

decisions that are serializable, trace-visible, and testable. Legacy [ROUTE: key]

tags are still accepted for older prompts.

Even though flows are programmatically structured, agents executing inside them must remain fully decoupled from the overall flow topology. Protolink achieves this through Dynamic Semantic Context Injection:

Topological Analysis: Before dispatching a step, the Flow orchestrator looks at the subsequent step in the topology (e.g., a single downstream agent, a parallel committee, or conditional routes).Context-Aware Prompt Generation: It automatically builds descriptive system instructions outlining the next target's capabilities, description, or routing choices (retrieved dynamically via theirAgentCard

from the Registry).Flow State Injection: It populates the prompt intotask.flow_state["prompt"]

.Execution: The executing agent's LLM automatically merges these instructions into its system prompt during inference. The agent adapts its behavior at runtime to optimize its output format for the downstream receiver without any hardcoded integration!

In simple terms, the Flow tells the agents what to do and what format to use for their output based on the downstream agents' capabilities. In the Pipeline example below, the orchestrator tells the researcher that its output will be used by the summarizer, so it should format its output in a way that the summarizer can understand. This is done automatically and transparently to the developer!

You can define and run structured flows programmatically using both asynchronous and synchronous APIs.

A Pipeline

runs a predefined sequence of agents, passing the output of one agent as the input to the next:

from protolink.flows import Pipeline
from protolink.models import Task

pipeline = Pipeline(
    steps=["researcher", "summarizer"],
    registry=registry  # Optional: allows discovering string-based agents by name
)

task = Task.create_infer(prompt="Research the future of Agentic computing.")
result = await pipeline.execute(task)

Pipelines support a fluid builder pattern for dynamic step configuration:

pipeline = Pipeline(registry=registry).add_step("researcher").add_step("summarizer")
result = await pipeline.execute(task)

For simple scripts, Jupyter Notebooks, or background workers where async syntax is not preferred, use the .sync

blocking wrapper:

result = pipeline.sync.execute(task)
print(result.get_last_part_content())

Since every flow primitive inherits from the base Flow

class, they are fully polymorphic. You can nest complex flow structures inside one another seamlessly:

from protolink.flows import Pipeline, Parallel

review_committee = Parallel(branches=["editor", "fact_checker"], registry=registry)

orchestrated_flow = Pipeline(registry=registry) \
    .add_step("researcher") \
    .add_step(review_committee) \
    .add_step("summarizer")

result = await orchestrated_flow.execute(task)

Protolink provides non-invasive, out-of-the-box observability for agent task execution, LLM inferences, and tool calls. It utilizes Python's contextvars

to automatically track execution states asynchronously without cluttering core method signatures.

Currently supported telemetry options:

LocalTraceTelemetry- built-in local traces for replay/debugging, with optional JSONL outputLangfuseLangSmith

For local debugging, use LocalTraceTelemetry

without any external service:

from protolink import Agent, AgentCard, LocalTraceTelemetry, Task, create_llm

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

agent = Agent(
    card=AgentCard(name="debug_agent", description="Local debug agent", url="runtime://debug"),
    llm=create_llm(
        "mock",
        sequential_responses=[
            {"type": "final", "content": "done"}
        ],
    ),
    telemetry=telemetry,
)

result = await agent.handle_task(Task.create_infer(prompt="Run a traced task"))
trace = telemetry.recorder.replay()[-1]

print(result.get_last_part_content())  # done
print(trace["trace_id"])
print([span["kind"] for span in trace["spans"]])  # task, llm, ...

Local traces capture task spans, LLM events, tool calls, retry counts, redacted payloads, and model metadata. This makes the infer loop easy to inspect before sending anything to Langfuse, LangSmith, or another observability backend.

For hosted telemetry, install the optional dependency (uv add "protolink[telemetry]"

) and inject the tracker to your agent. You can also broadcast events to multiple trackers using MultiTelemetry

:

from protolink.telemetry import LangfuseTelemetry, LangSmithTelemetry, MultiTelemetry
from protolink.agents import Agent

agent = Agent(
    card={"name": "ObserverAgent", "url": "http://127.0.0.1:8000"},
    telemetry=MultiTelemetry([LangfuseTelemetry(), LangSmithTelemetry()])
)

See the Telemetry Documentation for full setup instructions and custom integrations.

Protolink provides a flexible logging system to track agent activity across different outputs.

ConsoleLogger: Provides colored, human-readable logs in the terminal.FileLogger: Persists logs to disk, with optionalJSON formatting for easy ingestion into log management tools.QuietLogger: Implements the logger interface while intentionally dropping every log message.BaseLogger: Abstract base class for creating custom logging implementations.

MIT

All contributions are more than welcome! Please see CONTRIBUTING.md for more information.

source & further reading

github.com — original article