Skip to main content

Agents

Agents are the core building blocks in Protolink.

Concepts​

An Agent is ProtoLink's A2A-first runtime entity. It owns an AgentCard, receives and returns Task objects composed of Message, Part, and Artifact primitives, and can act as both client and server. It is the core building block of Protolink, responsible for managing identity, capabilities, and interactions between agents. The Agent integrates key components such as tools, LLMs, knowledge and RAG, transport, state, storage, telemetry, and logging.

Agent(..., transport="http", a2a=True) adds A2A 1.0 inbound and outbound translation without changing handle_task(Task) or removing the native ProtoLink endpoints. The default a2a=False preserves the previous native-only behavior.

Agents communicate through Tasks, the fundamental unit of work:

  • Receive tasks via handle_task()
  • Send tasks via call_agent()

Agents also register themselves and discover others via the registry. For inference requests ("infer"), ProtoLink automatically manages the full LLM interaction cycle until the task is resolved.

Each component is pluggable to the agent and can be replaced with your own implementation.

High‑level ideas:

  • A2A primitives: cards describe agents, tasks carry work, and messages, parts, and artifacts carry instructions and results.
  • Unified model: a single Agent instance can send and receive messages.
  • AgentCard: a small model describing the agent (name, description, metadata).
  • Modules:
    • LLMs (e.g. OpenAILLM, AnthropicLLM, LlamaCPPLocalLLM, LlamaCPPServerLLM, OllamaLLM).
    • Tools (native Python functions or MCP‑backed tools).
    • Storage (e.g. InMemoryStorage, SQLiteStorage).
    • Telemetry (e.g. LocalTraceTelemetry, LangfuseTelemetry, LangSmithTelemetry).
    • Logger (e.g. ConsoleLogger, FileLogger, QuietLogger).
  • Transport abstraction: agents communicate over transports such as HTTP, SSE JSON-RPC, WebSocket, gRPC, or the in-process runtime transport.
Agent Architecture

Creating an Agent​

A minimal agent consists of three pieces:

  1. An AgentCard describing the agent.
  2. A Transport implementation.
  3. An optional LLM and tools.

Example:

from protolink.agents import Agent
from protolink.models import AgentCard
from protolink.transport import HTTPTransport
from protolink.llms.api import OpenAILLM

# Agent card can be an AgentCard object or a dict for simplicity, both are handled the same way.
# Option 1: Using AgentCard object
agent_card = AgentCard(
    name="example_agent",
    description="A dummy agent",
    url="http://localhost:8000",
)

# Option 2: Using dictionary (simpler)
card_dict = {
    "name": "example_agent",
    "description": "A dummy agent",
    "url": "http://localhost:8000"
}

transport = HTTPTransport(url="http://localhost:8000")
llm = OpenAILLM(model="gpt-4o-mini")

# Both approaches work
agent = Agent(card=agent_card, transport=transport, llm=llm)
# OR
agent = Agent(card=card_dict, transport=transport, llm=llm)

You can then attach tools and start the agent.

When served through an HTTP-compatible transport ("http", "sse", "json-rpc", or "sse-json-rpc"), the Agent and Registry also expose browser pages for local inspection:

  • Registry status: GET <registry-url>/status
  • Agent status: GET <agent-url>/status
  • Agent chat: GET <agent-url>/chat for a browser UI, plus POST <agent-url>/chat for chat messages when the agent has an LLM

These pages are generated by the built-in status and chat renderers and are mounted from the same server endpoint specs as the JSON APIs. WebSocketTransport and RuntimeTransport keep the same logical endpoint definitions for Protolink clients, but they do not serve normal browser HTML pages directly.

Inspect registered agents in Devtools

If the agent is registered, run protolink dashboard --registry-url http://localhost:9000 --open to inspect its card, health, and chat support.

Registry status page
Agent status page
Agent chat page

Agent-to-Agent Communication​

Agents communicate over a chosen transport.

Common patterns:

  • RuntimeTransport: agents operate dedicated native local transports connected via a globally shared memory registry. This mirrors distributed HTTP environments perfectly, enabling zero network overhead testing workflows while retaining accurate boundaries.
  • HTTPTransport / SSEJSONRPCTransport: agents expose normal HTTP endpoints so other agents, CLIs, dashboards, browser pages, or external clients can send requests. SSE adds streamed task events over text/event-stream.
  • WebSocketTransport: agents expose the same endpoint specs over JSON frames on a WebSocket connection. Use this for streamed task events and long-lived clients rather than direct browser page URLs.

Agent Transport Layers​

LayerResponsibility
AgentDomain logic (what to do with a Task)
AgentServerWiring & lifecycle (server orchestration)
TransportProtocol abstraction (HTTP, SSE, WS, runtime)
BackendFramework-specific routing (Starlette/FastAPI)

e.g.

Agent.handle_task() -> AgentServer -> Transport.setup_routes() -> Backend creates route

Agent API Reference​

This section provides a detailed API reference for the Agent base class in protolink.agents.base. It is the core component for creating pluggable, A2A-based agents while combining client, server, execution, and runtime modules in one facade. HTTP agents can opt into the dedicated A2A 1.0 adapters described in A2A compatibility.

Unified Agent Model

Protolink's Agent combines client and server functionality in a single class. You can send tasks and messages to peers while also serving incoming requests; protocol-specific translation remains at the server boundary.

Core runtime moduleAgent

The public facade for identity, lifecycle, transport wiring, task execution, tools, knowledge retrieval, LLM inference, state, policy, telemetry, and registry discovery.

protolink.agents.Agent
Unified client and serverAsync and sync facadesTask, message, and stream APIsAutomatic knowledge toolsPolicy and approval checkpointsPluggable modules
ConstructBind card metadata to transport, registry, LLM, state, storage, telemetry, logging, and policy dependencies.Agent(card, transport, llm)
RunStart, stop, register, heartbeat, and expose the server endpoints owned by the agent runtime.start() / stop()
ExecuteHandle incoming tasks, retrieve knowledge, invoke tools, run inference loops, stream events, and emit durable run snapshots.handle_task()
CoordinateCall other agents, discover peers, inspect state, cancel work, and compact conversation history.call_agent()

Implementation Layout​

protolink.agents.base.Agent is the stable public facade. Internally, the agent package keeps the constructor and dependency wiring in base.py, the core task and LLM execution loop in engine.py, reusable behavior chunks in mixins.py, state-request normalization in helpers.py, and the blocking convenience facade in sync.py.

Constructor​

Agent​

classprotolink.agents.Agent
source
Agent(
  card: AgentCard | dict[str, Any],
  transport: TransportType | Transport | None = None,
  registry: TransportType | Registry | RegistryClient | None = None,
  registry_url: str | None = None,
  llm: LLM | None = None,
  system_prompt: str | None = None,
  storage: Storage | None = None,
  state: list[StateMode] | State | None = None,
  telemetry: Telemetry | None = None,
  skills: Literal["auto", "fixed"] = "auto",
  logger: BaseLogger | None = None,
  discovery_ttl: int = 0,
  *,
  override_system_prompt: bool = False,
  verbosity: Literal[0, 1, 2] = 1,
  expose_chat: bool = True,
  a2a: bool = False,
  authenticator: Authenticator | None = None,
  credentials: str | None = None,
  policy: Policy | None = None,
  approval_handler: ApprovalHandlerLike | None = None,
  run_store: Any | None = None,
  registry_heartbeat_interval: float | None = None,
  knowledge: Knowledge | Retriever | Sequence[Knowledge | Retriever] | None = None,
  retrieval: Literal["auto", "always", "required"] = "auto",
)

Create the stable Agent facade and wire its identity, execution engine, communication clients, server routes, state, tools, policy, and observability dependencies. Construction does not start a server or register the card; call start() or register() explicitly when those side effects are wanted.

Parameters

cardAgentCard | dict[str, Any]required

Identity and capability metadata for this Agent. Dictionaries are normalized with AgentCard.from_dict(). The card URL is also used when ProtoLink must construct a transport from a short alias, so it must match the address or runtime URI at which peers can reach the Agent.

transportTransportType | Transport | Nonedefault: None

Inbound and outbound communication layer. A registered alias such as "http", "runtime", "websocket", or "grpc" creates a transport with defaults derived from card.url. A concrete instance preserves its TLS, retry, limits, keepalive, metrics, and ownership configuration. None creates a local facade with no client or server.

registryTransportType | Registry | RegistryClient | Nonedefault: None

Optional discovery connection. A Registry contributes its client, a RegistryClient is used directly, and a transport alias creates a default client at registry_url. Without one, discovery returns an empty list and registration methods are no-ops.

registry_urlstr | Nonedefault: None

Registry address used only when registry is a transport alias. Put advanced TLS and capacity settings on a configured registry transport and pass its RegistryClient instead.

llmLLM | Nonedefault: None

Optional language model used for explicit infer parts. Assignment calls llm.validate_connection() and uses its result to update card.capabilities.has_llm. Depending on the adapter, validation may contact a provider or local server during Agent construction.

system_promptstr | Nonedefault: None

Agent-specific role and behavior instructions appended to ProtoLink's runtime prompt. Tool, delegation, flow, and action instructions are compiled separately. Set override_system_prompt=True only when the supplied text should replace that built-in blueprint.

storageStorage | Nonedefault: None

Persistence backend shared by the Agent and its State object. None creates an InMemoryStorage namespace based on card.name.

statelist[StateMode] | State | Nonedefault: None

Persistent-state configuration. A list enables selected stores such as "conversation", "tools", "task", and "flow"; a State instance is adopted directly. None is intentionally stateless even though an in-memory storage object still exists.

telemetryTelemetry | Nonedefault: None

Observer receiving task, tool, LLM, and inference events. The setter binds the telemetry object back to this Agent.

skillsLiteral["auto", "fixed"]default: "auto"

"auto" advertises skills inferred from registered tools while retaining card-defined skills. "fixed" leaves the card's declared skill list under application control.

loggerBaseLogger | Nonedefault: None

Logging implementation. When omitted, ProtoLink creates a namespaced ConsoleLogger whose level follows verbosity.

discovery_ttlintdefault: 0

Seconds to cache registry discovery results per filter. Zero disables caching, so every discovery request reaches the registry.

override_system_promptbooldefault: False

Replace the generated runtime prompt with system_prompt instead of treating it as complementary instructions. This can remove built-in action guidance, so use it only when the replacement prompt defines the complete contract.

verbosityLiteral[0, 1, 2]default: 1

Default Agent log level: 0 suppresses ordinary Agent logs, 1 emits informational lifecycle messages, and 2 enables debug detail. A supplied logger owns its own level.

expose_chatbooldefault: True

Allow the built-in chat handler and browser page when an LLM and HTTP-compatible server are available. It does not create an LLM or transport.

a2abooldefault: False

Enable the A2A 1.0 translation boundary. The current setter requires the exact HTTP transport; native ProtoLink endpoints remain available. Agent-originated A2A calls enforce same-origin advertised interfaces.

authenticatorAuthenticator | Nonedefault: None

Verifier for incoming transport requests. Authentication is enforced at the server boundary before task execution.

credentialsstr | Nonedefault: None

Credential value attached by the outbound client. Treat serialized configurations containing it as sensitive.

policyPolicy | Nonedefault: None

Runtime policy evaluated before tools, state mutation, history compaction, and other concrete actions. None installs an allow-by-default CapabilityPolicy, while tool- or run-level capability restrictions can still narrow access.

approval_handlerApprovalHandlerLike | Nonedefault: None

Synchronous or asynchronous application callback used to resolve typed approval checkpoints requested by policy.

run_storeAny | Nonedefault: None

Optional object implementing the run-store protocol. The engine writes terminal and streamed task snapshots to it; the store is observational and does not replace the process-local active-task registry.

registry_heartbeat_intervalfloat | Nonedefault: None

Seconds between heartbeats after successful registration. None disables the loop. Values below 0.1 are clamped to 0.1 seconds.

knowledgeKnowledge | Retriever | Sequence[Knowledge | Retriever] | Nonedefault: None

One knowledge source, a structural retriever, or a sequence of sources. Each source becomes a typed search_<name> tool available to the inference loop. Plain retrievers are wrapped as knowledge named "knowledge"; wrap them in Knowledge to provide a specific name, description, result limit, or reranker.

retrievalLiteral["auto", "always", "required"]default: "auto"

Default retrieval behavior for infer tasks. "auto" lets the model choose a knowledge tool, "always" retrieves before the first model call, and "required" additionally raises KnowledgeNotFoundError when no selected source can provide a usable passage inside the bounded model context. Per-task metadata may strengthen this mode but cannot weaken it.

Construction side effects

The constructor creates default storage, logger, policy, state, and sync facades and may construct transport clients and routes. LLM connection validation can perform I/O. Attaching knowledge creates retrieval tools and capability metadata, but staged knowledge sources remain lazy until ready() or the first search. Construction does not bind Agent server ports, register the card, or begin heartbeats until lifecycle methods run.

from protolink.agents import Agent
from protolink.models import AgentCard
from protolink.transport import HTTPTransport
from protolink.llms.api import OpenAILLM

url = "http://localhost:8020"
card = AgentCard(name="my_agent", description="Example agent", url=url)
llm = OpenAILLM(model="gpt-4")
transport = HTTPTransport(url=url)

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

Simple and Advanced Transports​

The Agent API uses progressive control. Pass a registered transport name when defaults are sufficient:

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

This is the prototyping path: ProtoLink creates an HTTPTransport from card.url, applies safe default limits, collects local metrics, and leaves retries disabled.

For TLS, resource policies, retries, or protocol-specific constructor options, build the transport explicitly:

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

card = AgentCard(
    name="production-agent",
    description="Production task worker",
    url="grpcs://agent.internal:9443",
)
transport_config = TransportConfig(
    limits=TransportLimits(
        max_request_bytes=8 * 1024 * 1024,
        max_response_bytes=8 * 1024 * 1024,
        max_concurrent_requests=200,
        max_concurrent_streams=50,
    ),
    retry=RetryPolicy(max_attempts=3),
    shutdown_timeout=10.0,
)
transport = GRPCTransport(
    url=card.url,
    tls=TLSConfig(
        certfile="certs/agent.pem",
        keyfile="certs/agent-key.pem",
        cafile="certs/ca.pem",
    ),
    config=transport_config,
)

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

Agent deliberately does not duplicate tls= or transport_config= arguments. TLS, limits, retries, keepalive, and connection ownership belong to the transport. This keeps the common Agent constructor small and lets the Agent transport and Registry transport use independent certificates and capacity policies.

An Agent uses its concrete transport in both directions: its server receives tasks from peers and its client sends tasks to peers. Inspect that instance through agent.transport; its config, capabilities, metrics, and health() surfaces are documented in the transport reference.

For an advanced Registry connection, construct its transport separately and wrap it in RegistryClient. Passing registry="http" remains the simple default path.

Durable Task Snapshots​

Agents remain stateless by default, but production services and CLIs often need a durable record of the task state that was returned to a user. Pass a RunStore implementation to run_store to persist snapshots without changing task execution code.

from protolink import Agent, AgentCard, SQLiteRunStore

store = SQLiteRunStore("runs.db")
agent = Agent(
    AgentCard(name="worker", description="Worker", url="runtime://worker"),
    llm=llm,
    run_store=store,
)

The built-in SQLiteRunStore indexes task ID, state, run ID, session ID, trace ID, and agent name. Applications can implement the same RunStore protocol for Postgres, object storage, or an existing application database. The store is observational: it records completed, failed, canceled, and streamed task snapshots, but active cancellation and live execution still use the process-local task registry.

Lifecycle Methods​

These methods control the agent's server component lifecycle.

Agent.start​

methodprotolink.agents.Agent.start
source
start(
  *,
  register: bool = True,
  background: bool = False,
) -> None

Start the configured server, optionally register the Agent, and keep its lifecycle alive. This is a synchronous entry point even though the underlying server and registry operations are asynchronous.

Parameters

registerbooldefault: True

Register card with the configured registry after server startup. If registration fails with a connection error, the server remains running but is not discoverable. With no registry client, this flag has no effect.

backgroundbooldefault: False

When False, run the lifecycle with asyncio.run() and block the calling thread. When True, create a non-daemon thread and a dedicated event loop, wait for startup readiness, then return to the caller.

Returns

NoneNone

Background mode returns after readiness or the ten-second startup wait. Blocking mode returns only after shutdown.

Raises

startup error

Server startup failures, including address conflicts, propagate to the caller. Background mode captures the exception in its thread and re-raises it after readiness synchronization.

Active event loops

Calling start(background=False) inside an already-running event loop blocks that loop. Use background=True from notebooks, ASGI applications, and other async hosts.

Agent.stop​

methodprotolink.agents.Agent.stop
source
stop() -> None

Request graceful shutdown of an Agent started in background mode and synchronously wait for its lifecycle thread to exit. Cancellation of the private lifecycle task triggers registry unregistration, heartbeat cleanup, and server shutdown.

Returns

NoneNone

Returns after the background thread exits or after the ten-second join timeout.

Idempotent teardown

Repeated calls are safe. stop() is designed around the background lifecycle; normal blocking mode is ordinarily stopped by interrupting the process lifecycle.

Execution Models & Lifecycle​

The start() method is the primary entrypoint for running an agent. To provide a "minimal boilerplate" experience, Protolink's lifecycle management automatically isolates the agent's internal async operations when running in the background, making it extremely robust across different environments.

The background Parameter​

The background parameter controls the execution mode and event loop isolation:

  • background=True: Starts the agent in a dedicated background thread with its own isolated asyncio event loop. It returns immediately. This is the recommended mode when running agents from Jupyter Notebooks, inside existing asyncio applications, or when orchestrating multiple agents in a single script.
  • background=False (Default): Takes over the main thread and blocks execution until the agent is stopped (e.g., via Ctrl+C in a terminal). Ideal for standalone agent processes.

Seamless Synchronous Teardown​

Because background=True isolates the agent in its own thread, shutting down the agent is incredibly simple. You just call agent.stop(). The stop() method operates synchronously, it tells the background thread to shut down and blocks for a fraction of a second to gracefully close Uvicorn and unregister from the registry. You do not need to await it, and it will never trigger messy CancelledError exceptions in your main event loop.

Common Usage Patterns​

1. Standalone Python Script For simple scripts where the agent is the main process, use the default blocking mode:

# This will take over the main thread and block until interrupted
agent.start()

2. Multi-Agent Script If you need to start multiple agents in a single script, run them in the background and gracefully stop them at the end.

agent_a.start(background=True)
agent_b.start(background=True)

# ... interact with agents ...

agent_a.stop()
agent_b.stop()

3. Jupyter Notebooks & Async Apps Jupyter Notebooks and async frameworks (like FastAPI) already have an active event loop. Using background=True safely isolates the agent from this loop:

async def main():
    agent.start(background=True)  # Spawns isolated thread, safe for async context
    
    # ... your async app logic ...
    
    agent.stop()  # Cleanly shuts down the thread without blocking your loop permanently
Graceful Shutdown

Always use agent.stop() to ensure that the agent unregisters from the registry and releases its network ports cleanly. In a standard script, agent.start(background=False) handles KeyboardInterrupt gracefully out of the box.

Transport Management​

Agent.transport​

propertyprotolink.agents.Agent.transport
source
transport: Transport | None

Read or replace the communication transport used for both outbound client calls and inbound server routes.

Setter value

transportTransportType | Transportrequired

A registered transport alias or configured transport instance. Setting an alias creates a transport from card.url. Setting a value rebuilds the associated AgentClient and AgentServer, then updates the card's transport and streaming capability metadata. Although the setter annotation includes None, the implementation rejects it with ValueError; construct the Agent without a transport instead of assigning None.

Lifecycle boundary

Reassigning this property configures objects and routes but does not start the new server. Avoid swapping transports while an Agent is serving; stop it first so connection ownership and open ports remain deterministic.

Agent.a2a​

read-only propertyprotolink.agents.Agent.a2a
source
a2a: bool

Report whether the optional A2A 1.0 compatibility boundary was enabled at construction. This is a configuration flag, not a live peer-capability probe.

Agent.client​

read-only propertyprotolink.agents.Agent.client
source
client: AgentClient | None

Return the outbound client built around the current transport. It is None when no transport is configured; high-level methods such as call_agent() raise rather than silently ignoring that condition.

Agent.server​

read-only propertyprotolink.agents.Agent.server
source
server: AgentServer | None

Return the inbound server facade created for the current transport, or None when that transport has no server implementation. Runtime and network transports can expose the same logical endpoint specifications through different backends.

Task and Message Handling​

Core Task Processing​

Agent.run_task​

async methodprotolink.agents.Agent.run_task
source
async run_task(task: Task) -> Task

Run the configured handle_task() implementation inside the live-execution registry. Server routes use this wrapper so even a completely overridden handler remains discoverable through active_task_ids and cancellable by task ID.

Parameters

taskTaskrequired
The mutable protocol task to execute. Terminal tasks are returned immediately without registering a new execution.

Returns

taskTask
The handler result. Successful, failed, and canceled snapshots are offered to run_store when configured.
Cancellation behavior
Protocol cancellation requested through cancel_task() is converted into a returned task in the canceled state. External coroutine cancellation is also persisted as canceled but asyncio.CancelledError is re-raised to its caller.

Agent.run_task_streaming​

async generatorprotolink.agents.Agent.run_task_streaming
source
run_task_streaming(task: Task) -> AsyncIterator[Any]

Stream the configured handler under the same active-task registration used by non-streaming server routes. This outer wrapper gives custom streaming handlers the runtime's cancellation and snapshot guarantees.

Parameters

taskTaskrequired
Task whose status and outputs are streamed. A terminal task yields one final status event and stops.

Yields

eventAny
Typed task status, progress, LLM, artifact, or error events produced by handle_task_streaming().
Final cancellation event
A successfully canceled stream ends with a final TaskStatusUpdateEvent whose state is canceled and whose metadata includes the serialized task and reason.
Abandoned consumers
If a consumer closes the iterator before a terminal event, unfinished work is marked canceled with a stream-closure reason and persisted before cleanup.

Agent.handle_task​

async methodprotolink.agents.Agent.handle_task
source
async handle_task(task: Task) -> Task

Provide the default task-handler boundary. It normalizes a RunContext, emits best-effort telemetry start/end hooks, and delegates deterministic execution to execute_task(). Telemetry failures are logged once per hook and cannot alter the task outcome. Override this method for application-specific routing or orchestration, not merely to add a tool or LLM.

Parameters

taskTaskrequired
Task to process. Only explicit executable parts are acted upon; ordinary native text does not implicitly trigger inference.

Returns

taskTask
Updated task returned by the execution engine or custom handler.
Subclassing
Remote cancellation wraps this method through run_task(). Direct callers of a fully custom handler should use that wrapper too, or call execute_task() inside the override to retain the standard engine.

Agent.handle_task_streaming​

async generatorprotolink.agents.Agent.handle_task_streaming
source
handle_task_streaming(task: Task) -> AsyncIterator[Any]

Execute a task while emitting its lifecycle as typed events. The default stream begins with a working-state transition, relays tool and LLM progress, emits artifact updates, and finishes with a terminal status containing the complete task.

Parameters

taskTaskrequired
Task to mutate as streamed work completes.

Yields

eventTaskStatusUpdateEvent | TaskProgressEvent | TaskLLMStreamEvent | TaskArtifactUpdateEvent | TaskErrorEvent
Provider-neutral events suitable for SSE, WebSocket, runtime, gRPC, or direct consumers.
Error contract
Execution failures are converted into a TaskErrorEvent followed by a final failed status event. Consumers should use the final status metadata as the authoritative task snapshot.
Internal-result privacy
For tool_result and agent_call_result events, client-visible TaskLLMStreamEvent.metadata retains correlation fields and sets result_omitted=true but does not carry the internal result. Ordinary tool observations can remain in private LLM history and configured telemetry. Generated knowledge-tool passages are stricter: raw evidence is available only to the active model loop, then replaced in persistent history and observability with an omission receipt and bounded search statistics.

Agent.execute_task​

async methodprotolink.agents.Agent.execute_task
source
async execute_task(task: Task) -> Task

Execute one deterministic step from the most recently appended message or artifact. tool_call parts invoke registered tools, infer parts enter the LLM loop, and authenticated inbound A2A text is translated to inference by the default engine. Other part types are left inert.

Parameters

taskTaskrequired
The same mutable Task instance receives outputs, artifacts, state transitions, and normalized run-context metadata.

Returns

taskTask
The input object after execution. Success completes it, error output fails it, and an explicit input-required status pauses it.
History isolation
The engine binds a task-local LLM history. Stateless tasks get fresh history; conversation-enabled sessions load and lock persisted history so concurrent work cannot interleave one session. Normally failed turns are discarded, but a failed turn with a new action_result receipt is retained because it contains the observation of a completed side effect.
Task-wide budgets
One task-scoped BudgetEnforcer is shared by explicit tool parts and every iteration or retry inside all infer parts. Inline nested tasks use their own scope and restore the parent budget afterward.
Completed-action receipts
Outputs are attached and offered to run_store after each completed top-level part. Successful tools or delegations selected inside LLM.infer() additionally create and immediately snapshot an Artifact(kind="action_result") JSON receipt with completion status, action ID, and source/kind/step metadata. Internal results are deliberately omitted from this client-visible artifact. Ordinary tool observations remain private to model history; generated knowledge evidence is also scrubbed from persistent history and telemetry after the active loop. Later failure, cancellation, or budget exhaustion therefore preserves evidence of completed side effects without exposing internal result data.

Agent.compact_history​

async methodprotolink.agents.Agent.compact_history
source
async compact_history(
  request: HistoryCompactionRequest | dict[str, Any] | None = None,
) -> HistoryCompactionResult

Compact the Agent LLM's current or persisted session history through the control plane. The operation is deliberately outside task inference, so it is never advertised to the model and consumes no inference step.

Parameters

requestHistoryCompactionRequest | dict[str, Any] | Nonedefault: None
Compaction strategy, limits, and optional session ID. None uses HistoryCompactionRequest defaults. Dictionaries are validated through from_dict().

Returns

resultHistoryCompactionResult
Before/after counts, estimated tokens, strategy, and summary metadata.

Raises

RuntimeError
The Agent has no LLM.
TypeError
The request or an authorized replacement payload has the wrong shape.
policy error
The llm.history.compact capability is denied or requires unavailable approval.

Agent.describe_state​

async methodprotolink.agents.Agent.describe_state
source
async describe_state(
  request: str | StateOperationRequest | dict[str, Any] | None = None,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
  include_data: bool | None = None,
) -> StateOperationResult

Inspect enabled persistent-state stores without mutating them. A string request is shorthand for a session ID; explicit keyword arguments override values supplied in a request object or dictionary.

Parameters

requeststr | StateOperationRequest | dict[str, Any] | Nonedefault: None
Optional normalized state-operation request or session-ID shorthand.
session_idstr | Nonedefault: None
Limit session-aware reports, especially conversation history, to one logical session.
storestuple[str, ...] | list[str] | Nonedefault: None
Store names to report. None asks the State implementation for its normal scope.
include_databool | Nonedefault: None
Whether reports may include stored data in addition to existence and counts; omitted preserves any request value.

Returns

resultStateOperationResult
A structured per-store report, including disabled or missing stores rather than hiding them.
Authorization
The operation requires state.describe. Although read-only, policy may still redact, replace, approve, or deny the concrete request.

Agent.reset_state​

async methodprotolink.agents.Agent.reset_state
source
async reset_state(
  request: str | StateOperationRequest | dict[str, Any] | None = None,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
) -> StateOperationResult

Clear persistent Agent state through an authorized control-plane action. Supplying a session ID precisely clears that conversation session; omitting it performs a namespace-level reset of all enabled stores.

Parameters

requeststr | StateOperationRequest | dict[str, Any] | Nonedefault: None
Request object, dictionary, session-ID shorthand, or default full-reset request.
session_idstr | Nonedefault: None
Conversation session to clear. This is the safer, narrower form for user-facing reset controls.
storestuple[str, ...] | list[str] | Nonedefault: None
Requested store selection. Partial namespace-wide resets may be rejected because the current storage abstraction resets by namespace.

Returns

resultStateOperationResult
Structured reset, missing-store, and error reports.
Destructive operation
The action requires state.reset authorization and mutates persisted data. Use a session_id whenever the intent is to forget one conversation rather than the Agent namespace.

Agent.compact_state​

async methodprotolink.agents.Agent.compact_state
source
async compact_state(
  request: str | StateOperationRequest | dict[str, Any] | None = None,
  *,
  session_id: str | None = None,
  strategy: HistoryCompactionStrategy | None = None,
  max_messages: int | None = None,
  max_tokens: int | None = None,
  preserve_recent: int | None = None,
  summary_max_tokens: int | None = None,
) -> StateOperationResult

Load one persisted conversation, compact it with the LLM-owned history compactor, save the replacement history, and return before/after state metadata.

Parameters

requeststr | StateOperationRequest | dict[str, Any] | Nonedefault: None
Base state-operation request or session shorthand.
session_idstr | Nonedefault: None
Required logical conversation session. If absent, the method returns a structured error result rather than raising.
strategyHistoryCompactionStrategy | Nonedefault: None
Recent-message, token-budget, or summary strategy. None preserves the request/default strategy.
max_messagesint | Nonedefault: None
Maximum retained messages for message-count compaction.
max_tokensint | Nonedefault: None
Estimated history-token budget for token compaction.
preserve_recentint | Nonedefault: None
Recent messages protected when older context is summarized.
summary_max_tokensint | Nonedefault: None
Maximum tokens requested for the generated summary.

Returns

resultStateOperationResult
A report showing whether conversation state existed, whether it was compacted, and nested compaction counts.
Requirements
Conversation state, an LLM, and an existing session are required. Missing prerequisites are reported in result.errors; policy denial still raises through the authorization layer.

Agent.cancel_task​

async methodprotolink.agents.Agent.cancel_task
source
async cancel_task(
  request: str | TaskCancellationRequest,
  reason: str | None = None,
) -> Task

Request best-effort cancellation of work currently registered on this process. The runtime marks both Task and RunContext cancellation state, signals the cooperative token, and interrupts the owning coroutine at its next await point.

Parameters

requeststr | TaskCancellationRequestrequired
Active task ID or typed request. A typed request can carry its own reason.
reasonstr | Nonedefault: None
Human-readable reason. When supplied alongside a typed request, this explicit value takes precedence.

Returns

taskTask
The active task after transition to canceled.

Raises

TaskNotFoundError
No active execution with that ID exists on this Agent.
TaskNotCancelableError
The task has already reached a terminal state.
Best effort
Already-issued external side effects and synchronous CPU work cannot be forcibly undone. Custom long-running handlers should call the live token's raise_if_cancelled() at safe checkpoints.

Agent.get_cancellation_token​

methodprotolink.agents.Agent.get_cancellation_token
source
get_cancellation_token(task_id: str) -> CancellationToken | None

Return the process-local cooperative token for an active task.

Parameters

task_idstrrequired
Task ID currently visible in active_task_ids.

Returns

tokenCancellationToken | None
Live token, or None after completion or before registration. Tokens are never serialized into protocol objects.

Agent.active_task_ids​

read-only propertyprotolink.agents.Agent.active_task_ids
source
active_task_ids: tuple[str, ...]

Return an immutable snapshot of task IDs currently registered for live execution. Completed tasks disappear immediately; query a configured run store for historical snapshots.

Agent.invoke​

async methodprotolink.agents.Agent.invoke
source
async invoke(
  message: str,
  part_type: Literal["tool_call", "infer"] = "infer",
  tool_name: str | None = None,
  tool_args: dict[str, Any] | None = None,
  session_id: str = "invocation_session_id",
) -> str

Create a one-step task, process it through handle_task(), and return only the final part content. This convenience API is useful for direct application calls but intentionally discards the richer Task envelope.

Parameters

messagestrrequired
User prompt for inference. In tool-call mode it is not used as the tool argument payload.
part_typeLiteral["tool_call", "infer"]default: "infer"
Choose an LLM inference part or an explicit registered-tool call.
tool_namestr | Nonedefault: None
Registered tool name in tool-call mode. Omission becomes an empty name and therefore produces a normal tool-not-found output.
tool_argsdict[str, Any] | Nonedefault: None
Keyword arguments encoded into the tool-call part.
session_idstrdefault: "invocation_session_id"
Conversation-state partition attached to task metadata. The stable default means sequential invocations share history when conversation state is enabled.

Returns

responsestr
Last part content, or "No response generated" when the task produced none.

Raises

ValueError
An unsupported part_type was supplied.
Task details
Use handle_task(), run_task(), or the client API when callers need task state, artifacts, run context, or structured error information.

Agent.ask​

async methodprotolink.agents.Agent.ask
source
async ask(
  question: str,
  *,
  knowledge: str | list[str] | tuple[str, ...] | None = None,
  k: int | None = None,
  where: dict[str, Any] | None = None,
  citations: bool = True,
  session_id: str = "ask_session_id",
) -> RAGAnswer

Run deterministic retrieve-then-answer through the Agent's normal task, policy, cancellation, telemetry, history, budget, and inference boundaries. Unlike invoke() in automatic mode, this method always searches before the first model call.

Parameters

questionstrrequired
Non-empty user question used as both the retrieval query and original inference request.
knowledgestr | list[str] | tuple[str, ...] | Nonedefault: None
Attached knowledge name or names. Omission searches every attached source.
kint | Nonedefault: None
Maximum hits per selected source. Omission uses each source's default_k.
wheredict[str, Any] | Nonedefault: None
Metadata filter passed to every selected source.
citationsbooldefault: True
Request bracketed evidence labels and retain structured Citation values. When false, hits are still returned but RAGAnswer.citations is empty.
session_idstrdefault: "ask_session_id"
Conversation-state partition attached to the generated task.

Returns

answerRAGAnswer
Final model text together with the original query, normalized hits, and optional structured citations.

Raises

RuntimeError
No knowledge source is attached.
ValueError | TypeError
The question, selected names, result count, or filters are invalid.
retrieval or inference error
Search, policy, budget, provider, and task failures propagate through their normal typed errors.
Complete RAG guide
See Retrieval-Augmented Generation for managed indexes, existing vector databases, custom retrievers, retrieval modes, citations, and lifecycle operations.

Task Lifecycle​

The default Agent implementation manages Task.state for you:

  1. Incoming non-terminal tasks move to TaskState.WORKING.
  2. Successful tool or LLM outputs move the task to TaskState.COMPLETED.
  3. Error parts, failed tool outputs, or raised exceptions move the task to TaskState.FAILED.
  4. Status parts requesting more input move the task to TaskState.INPUT_REQUIRED.

Every successful state change is appended to task.metadata["state_history"]. Streaming handlers emit matching TaskStatusUpdateEvent events and include the final serialized task in the final status event metadata.

Before execution, the default runtime also normalizes RunContext into task.metadata["run_context"]. This gives applications one typed place for session IDs, trace IDs, workspace URIs, permission metadata, budgets, cancellation state, and parent/child agent chains. See Runtime for the full context and event-sink API.

AgentServer routes call run_task() and run_task_streaming(), so fully overridden handlers still receive active-task registration and remote cancellation. Direct application code should also use these wrappers when it invokes a fully custom handler. Prefer calling await self.execute_task(task) inside custom handlers when you only need to wrap or augment the default execution.

Live Task Cancellation​

await agent.cancel_task(task.id, reason="Stopped by user") controls work that is currently active on this Agent. It is different from Task.cancel(): the task helper records serializable lifecycle state, while the Agent API also signals a live CancellationToken and cancels the owning asyncio.Task.

The default LLM, tool, streaming, and delegation paths already check this token. A custom CPU loop can retrieve it and add explicit checkpoints:

async def handle_task(self, task: Task) -> Task:
    token = self.get_cancellation_token(task.id)
    for item in application_items:
        if token is not None:
            token.raise_if_cancelled()
        await process(item)
    return task.complete("done")

Only active runs appear in active_task_ids. Wait until task acceptance or a first status event before canceling; completed entries are removed and belong in application storage. Cancellation is best-effort because synchronous functions and external systems may not stop immediately. See Runtime cancellation for the complete contract.

The Inference Loop Integration​

When execute_task() encounters an infer part, it delegates to LLM.infer() with:

  1. The query: Extracted from the task's message content
  2. Knowledge retrieval: Deterministic pre-retrieval for "always" or "required" (with knowledge tools suppressed afterward to avoid duplicate reads), while "auto" exposes each source as an ordinary search tool
  3. The agent's tools: All registered tools passed as a dictionary
  4. Validated discovered Agents: Registry-advertised targets, excluding this Agent and names already in the ancestor chain
  5. An agent callback: Enables delegation only when at least one valid target was discovered
  6. Runtime controls: Policy authorization, cancellation, the normalized RunContext, and the task-scoped budget enforcer
  7. Optional stream observers: Used by handle_task_streaming() to emit task_llm_stream events while the final infer_output part is produced
# Simplified view of what happens inside execute_task()
result = await self.llm.infer(
    query=query,
    tools=self.tools,
    agent_callback=dispatch_discovered_agent,
    cancellation_token=token,
    run_context=context,
    budget_enforcer=task_budget,
)

Registry discovery is a best-effort prompt affordance. If discovery fails, local LLM and tool work continues without delegation targets. Existing custom LLM.infer() overrides keep working: Agent supplies budget_enforcer only when the override explicitly accepts it or **kwargs.

The agent callback (_handle_agent_call) is invoked when the LLM produces an agent_call action. It:

  1. Requires a Registry-advertised name and rejects a model-produced URL
  2. Rejects self-delegation and ancestor-cycle targets before dispatch
  3. Resolves the agent name to URL by querying the registry
  4. Creates a child Task and RunContext with the appropriate infer or tool-call part
  5. Sends the task to the target agent via call_agent()
  6. Validates the remote Task state and returns output only from a genuinely completed task. Item IDs distinguish new output from the outbound request, supporting both full-task and response-only transport shapes. Remote failed, canceled, non-terminal, empty-completion, and input-required responses are propagated explicitly instead of echoing the child request as a successful result.

This enables a coordinator agent to delegate work to specialized agents without manual orchestration.

Agent Delegation Flow
User Query → Coordinator Agent → LLM.infer()
                                    ↓
                               agent_call action
                                    ↓
                           _handle_agent_call()
                                    ↓
                           resolve agent URL
                                    ↓
                           call_agent(weather_agent)
                                    ↓
                           Weather Agent processes task
                                    ↓
                           Result returned to LLM
                                    ↓
                            LLM produces final response

Controlling Agent Delegation​

By default, any agent with an LLM can dynamically delegate work to other agents discovered via the registry. However, you can explicitly disable or control delegation using the agent's Capabilities:

  • delegation: A boolean flag (defaulting to True) indicating whether the agent is allowed to delegate tasks to other agents.
  • has_llm: A boolean flag (defaulting to False) showing whether the agent has an LLM as a core component.

Disabling Delegation​

If you set "delegation": False within the agent's card capabilities:

  1. The agent will not query the registry to discover other agents.
  2. The agent's prompt builder will not inject other agents' definitions or descriptions into the LLM system instructions.
  3. The inference engine's agent_callback is set to None, completely preventing any remote task delegation loops.

Configuration Example​

To disable delegation, simply define it in your AgentCard capabilities dictionary:

from protolink.agents import Agent

writer = Agent(
    card={
        "name": "writer",
        "url": "http://localhost:8051",
        "description": "Writes drafts and decides routes.",
        "capabilities": {
            "delegation": False  # Disables A2A delegation completely
        }
    },
    llm=llm,
)

Communication Methods​

Agent.call_agent​

async methodprotolink.agents.Agent.call_agent
source
async call_agent(
  agent_url: str,
  task: Task,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Send a complete Task to a peer through this Agent's configured client. Before dispatch, ProtoLink ensures that the task carries a RunContext and extends its agent chain for trace and delegation correlation.

Parameters

agent_urlstrrequired
Reachable peer URL or runtime URI.
taskTaskrequired
Mutable Task envelope to send. Transport serialization does not strip its native metadata when ProtoLink protocol is used.
protocolLiteral["auto", "protolink", "a2a"]default: "auto"
"auto" preserves the richer native contract and discovers A2A-only peers when A2A is enabled. The other values force one boundary.

Returns

taskTask
Peer response task with its updated lifecycle and outputs.

Raises

RuntimeError
This Agent has no configured transport/client.
transport or protocol error
Connection, authentication, translation, and peer errors propagate from AgentClient.

Agent.send_message_to​

async methodprotolink.agents.Agent.send_message_to
source
async send_message_to(
  agent_url: str,
  message: Message,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Message

Send a standalone Message to a peer and return its response message. Use call_agent() when lifecycle state, artifacts, metadata, cancellation, or structured-flow context matters.

Parameters

agent_urlstrrequired
Reachable peer address.
messageMessagerequired
Role and parts to send.
protocolLiteral["auto", "protolink", "a2a"]default: "auto"
Same protocol-selection rules as call_agent().

Returns

messageMessage
Response decoded from the peer.

Raises

RuntimeError
No transport/client is configured.
agent = Agent(card=card, transport="http", a2a=True)

# Preserve the richer native contract when the peer supports it; otherwise use A2A.
result = await agent.call_agent(peer_url, task, protocol="auto")

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

Agent-originated A2A calls require the advertised JSON-RPC interface to share the discovered Agent Card's origin. This secure policy is intentionally fixed on the compact Agent facade. Applications that explicitly trust a split-origin deployment can construct an AgentClient(..., a2a_allow_cross_origin=True) for that outbound integration.

At the A2A boundary, a standard user text part remains a ProtoLink Part(type="text") for custom handlers. The default Agent engine recognizes task.metadata["a2a_inbound"] and treats that text as an inference request when an LLM is configured. A ProtoLink infer prompt becomes standard A2A user text outbound. Standard text, data, file, and URI content can be translated. ProtoLink-specific tool_call parts, structured-flow state, runtime context, and native control endpoints are not portable A2A contracts; keep protocol="protolink" when a peer needs those details.

Synchronous API (SyncAgent)​

Protolink is built on an asynchronous foundation using asyncio, which is essential for handling concurrent agent interactions and streaming responses. However, many development workflows, such as data science notebooks, CLI tools, and simple automation scripts, benefit from a straightforward, blocking API.

The Agent class provides a .sync property, which is an instance of SyncAgent. This class acts as a thin, synchronous wrapper around the agent's core async methods.

Why Use the Sync API?​

  1. Reduced Boilerplate: Eliminates the need for async/await and event loop management in scripts.
  2. Environment Compatibility: Works seamlessly in standard Python environments and legacy codebases that are not yet async-ready.
  3. Prototyping: Allows for faster iteration when building simple "input-output" agent flows.

How it Works Internally​

The SyncAgent class does not re-implement any logic. Instead, it delegates calls to the agent's async methods using asyncio.run(). This ensures that all behavior, including tool execution, state management, and LLM orchestration, remains identical across both APIs.

Event Loop Conflict

The synchronous API is not thread-safe if called from within an active event loop (e.g., inside a FastAPI endpoint or an async function). Doing so will raise a RuntimeError. For async applications, always use the standard await agent.invoke() methods.

Key Sync Methods​

The sync facade exposes blocking equivalents with the same parameter and return contracts:

SyncAgent.invoke​

methodprotolink.agents.SyncAgent.invoke
source
invoke(
  message: str,
  part_type: Literal["tool_call", "infer"] = "infer",
  tool_name: str | None = None,
  tool_args: dict[str, Any] | None = None,
  session_id: str = "invocation_session_id",
) -> str

Blocking form of Agent.invoke(). Every argument and the returned final-part text have the same meaning; the wrapper runs the coroutine with asyncio.run().

Parameters

messagestrrequired
User prompt used to create an infer Part. In tool-call mode the wrapper still forwards it, but the underlying Agent builds the tool Part from tool_name and tool_args.
part_typeLiteral["tool_call", "infer"]default: "infer"
Select direct inference or an explicit tool call. Other values are rejected by Agent.invoke().
tool_namestr | Nonedefault: None
Registered tool name for tool-call mode. A falsey value becomes an empty tool name and produces the Agent's normal tool-not-found result.
tool_argsdict[str, Any] | Nonedefault: None
Keyword arguments placed in the generated tool-call Part. None and an empty mapping are normalized to an empty argument mapping.
session_idstrdefault: "invocation_session_id"
Session identifier written to task metadata before execution. The stable default shares conversation state across sequential invocations when conversation persistence is enabled.

SyncAgent.ask​

methodprotolink.agents.SyncAgent.ask
source
ask(
  question: str,
  *,
  knowledge: str | list[str] | tuple[str, ...] | None = None,
  k: int | None = None,
  where: dict[str, Any] | None = None,
  citations: bool = True,
  session_id: str = "ask_session_id",
) -> RAGAnswer

Blocking form of Agent.ask() with the same retrieval, filter, citation, and return contract. The wrapper uses asyncio.run() and must not be called from an active event loop.

SyncAgent.discover_agents​

methodprotolink.agents.SyncAgent.discover_agents
source
discover_agents(
  filter_by: dict[str, Any] | None = None,
) -> list[AgentCard]

Blocking registry discovery with the Agent's normal TTL cache and empty-list behavior when no registry is configured.

Parameters

filter_bydict[str, Any] | Nonedefault: None
Optional nested card-field criteria forwarded to registry discovery.

SyncAgent.call_agent​

methodprotolink.agents.SyncAgent.call_agent
source
call_agent(
  agent_url: str,
  task: Task,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Blocking form of Agent.call_agent(), including context propagation and protocol selection.

Parameters

agent_urlstrrequired
Reachable peer URL or runtime URI forwarded unchanged to the Agent client.
taskTaskrequired
Mutable task envelope sent to the peer after the Agent ensures its run context.
protocolLiteral["auto", "protolink", "a2a"]default: "auto"
Keyword-only protocol selector. "auto" prefers the native ProtoLink contract and discovers an A2A-only boundary when configured; the other values force one protocol.

SyncAgent.cancel_task​

methodprotolink.agents.SyncAgent.cancel_task
source
cancel_task(
  task_id: str,
  reason: str | None = None,
) -> Task

Blocking form of local live-task cancellation.

Parameters

task_idstrrequired
ID of an execution currently registered on the wrapped Agent. Unlike the async method, this facade does not accept a TaskCancellationRequest.
reasonstr | Nonedefault: None
Optional human-readable cancellation reason propagated to the task state and serialized RunContext.
Concurrency requirement
The target task must already be running on another coroutine or the Agent's background loop. A synchronous caller cannot interrupt work while blocked inside the same call stack.

Usage Example​

from protolink.agents import Agent
from protolink.models import Task

agent = Agent(
    card={"name": "my-agent", "description": "Runtime demo agent", "url": "runtime://agent"},
    transport="runtime",
)

# Use the .sync property for blocking calls
response = agent.sync.invoke("Hello, agent!")
print(f"Agent said: {response}")

# Discovering other agents synchronously
discovered = agent.sync.discover_agents(filter_by={"name": "weather-agent"})
if discovered:
    target_url = discovered[0].url
    # Call agent synchronously
    task = Task.create_infer("What is the temperature?")
    result = agent.sync.call_agent(target_url, task)

Skills Management​

Skills represent the capabilities that an agent can perform. Skills are stored in the AgentCard and can be automatically detected or added.

Skills Modes​

ModeDescription
"auto"Automatically detects skills from tools and public methods, and adds them to the AgentCard
"fixed"Uses only the skills explicitly defined in the AgentCard

Skill Detection​

When using "auto" mode, the agent detects skills from:

  1. Tools: Each registered tool becomes a skill when it is added.
  2. Card declarations: Existing AgentCard.skills are retained.

The internal detector can describe public methods, but the public Agent(..., skills="auto") path intentionally calls it with public-method detection disabled. This keeps infrastructure methods such as lifecycle and registry controls out of the advertised skill list.

# Auto-detect skills from tools only
agent = Agent(card, skills="auto")

# Use only skills defined in AgentCard
agent = Agent(card, skills="fixed")

Skills in AgentCard​

Skills are persisted in the AgentCard and serialized when the card is exported to JSON:

from protolink.models import AgentCard, AgentSkill

# Create skills manually in AgentCard
card = AgentCard(
    name="weather_agent",
    description="Weather information agent",
    skills=[
        AgentSkill(
            id="get_weather",
            description="Get current weather for a location",
            tags=["weather", "forecast"],
            examples=["What's the weather in New York?"]
        )
    ]
)

# Use fixed mode to only use these skills
agent = Agent(card, skills="fixed")

Knowledge Management​

Knowledge sources are specialized read-only tools. Attaching one keeps agent.knowledge, agent.tools, advertised skills, and Agent Card capability metadata synchronized. See the complete Retrieval-Augmented Generation guide for ingestion and retrieval.

Agent.add_knowledge​

methodprotolink.agents.Agent.add_knowledge
source
add_knowledge(
  knowledge: Knowledge | Retriever,
) -> Knowledge

Attach one knowledge source and register its generated search_<name> tool.

Parameters

knowledgeKnowledge | Retrieverrequired
A configured facade or structural retriever. A plain retriever is wrapped with the default name "knowledge".

Returns

knowledgeKnowledge
The normalized attached facade.

Raises

ValueError
The knowledge name is already attached or its generated tool name conflicts with an existing tool.
Capability metadata
Successful attachment sets both card.capabilities.rag and card.capabilities.tool_calling to true.

Agent.retriever​

decorator factoryprotolink.agents.Agent.retriever
source
retriever(
  *,
  name: str = "knowledge",
  description: str | None = None,
  default_k: int = 5,
)

Adapt and attach a synchronous or asynchronous application search function as retrieval-only knowledge. The callable receives query and, when its signature accepts them, k and where.

Tool Management​

Tools give agents explicit callable capabilities. ProtoLink supports opt-in built-ins, native Python functions, custom BaseTool implementations, and MCP adapters.

Agent.add_tool​

methodprotolink.agents.Agent.add_tool
source
add_tool(tool: BaseTool) -> None

Register or replace a runtime tool by name and synchronize its public skill advertisement.

Parameters

toolBaseToolrequired
Executable tool carrying a stable name, description, schemas, tags, and examples. Replacing an existing runtime tool also replaces its generated skill; an independently card-defined skill with the same ID is preserved on the first registration.
No execution
Registration has no external side effect beyond mutating agent.tools and card.skills. Policy, validation, approvals, telemetry, and cancellation run only when the tool is called.

Agent.tool​

decorator factoryprotolink.agents.Agent.tool
source
tool(
  name: str,
  description: str,
  input_schema: dict[str, Any] | None = None,
  output_schema: dict[str, Any] | None = None,
  tags: list[str] | None = None,
  examples: list[Any] | None = None,
  capabilities: list[str] | tuple[str, ...] | set[str] | None = None,
  action_builder: ActionBuilder | None = None,
)

Wrap a Python callable as a ProtoLink Tool, register it immediately, and return the original callable so ordinary direct Python usage remains possible.

Parameters

namestrrequired
Stable identifier exposed to models, peers, policy, and serialized configuration.
descriptionstrrequired
Purpose statement used in prompts and skill discovery.
input_schemadict[str, Any] | Nonedefault: None
Optional JSON Schema used to validate keyword arguments before authorization.
output_schemadict[str, Any] | Nonedefault: None
Descriptive return schema advertised with the tool.
tagslist[str] | Nonedefault: None
Discovery and presentation labels.
exampleslist[Any] | Nonedefault: None
Representative invocations copied to the generated Agent skill.
capabilitieslist[str] | tuple[str, ...] | set[str] | Nonedefault: None
Permission capabilities that policy must authorize immediately before execution.
action_builderActionBuilder | Nonedefault: None
Hook that can enrich the concrete RunAction with preview artifacts or metadata before approval.

Returns

decoratorCallable
Decorator that registers the wrapped function and returns that same function.

Agent.call_tool​

async methodprotolink.agents.Agent.call_tool
source
async call_tool(
  tool_name: str,
  **kwargs,
) -> Any

Validate, authorize, and execute a registered tool with a fresh RunContext associated with this Agent.

Parameters

tool_namestrrequired
Registered key in agent.tools.
**kwargsAny
Arguments validated against the tool schema, then passed to its callable after policy authorization.

Returns

resultAny
Raw tool result; unlike task execution, this method does not wrap success or failure in a tool_output Part.

Raises

ValueError
The named tool is not registered.
validation, policy, approval, or tool error
Direct calls propagate these errors to the caller.

Agent.call_tool_in_context​

async methodprotolink.agents.Agent.call_tool_in_context
source
async call_tool_in_context(
  tool_name: str,
  context: RunContext,
  **kwargs: Any,
) -> Any

Execute a tool while preserving an application-supplied RunContext. Use this form in deterministic flows and custom runtimes so permissions, trace IDs, budgets, workspace metadata, and cancellation state participate in authorization.

Parameters

tool_namestrrequired
Registered key in agent.tools. An unknown name raises ValueError before authorization.
contextRunContextrequired
Existing typed run context supplied unchanged to tool-action preparation and policy authorization.
**kwargsAny
Tool keyword arguments. The runtime validates supported schemas, authorizes the prepared action, and invokes the tool with the authorized argument mapping, which may differ from the original mapping.

Agent.authorize_action​

async methodprotolink.agents.Agent.authorize_action
source
async authorize_action(
  action: RunAction,
  context: RunContext | None = None,
) -> ActionAuthorization

Evaluate a fully prepared runtime action without executing its side effect. Custom orchestration can use the same policy and approval checkpoint as built-in tools.

Parameters

actionRunActionrequired
Concrete operation, payload, capabilities, description, and preview artifacts to evaluate.
contextRunContext | Nonedefault: None
Active run context; omission creates a fresh context containing this Agent in its chain.

Returns

authorizationActionAuthorization
Approved action, potentially with policy- or approver-modified payload.
Important
Execute authorization.action, not an earlier copy of the action. Policies and approvers may narrow or replace arguments during authorization.
# Using the decorator approach
@agent.tool("calculate", "Performs basic calculations")
def calculate(operation: str, a: float, b: float) -> float:
    if operation == "add":
        return a + b
    elif operation == "multiply":
        return a * b
    else:
        raise ValueError(f"Unsupported operation: {operation}")

# Direct registration of built-in Tool instances
from protolink.tools import current_datetime, web_search

agent.add_tool(current_datetime())
agent.add_tool(web_search())  # Brave by default; calls may select engine="duckduckgo".

Built-ins are never enabled automatically. Registered built-ins follow the same validation, policy, telemetry, cancellation, and skill-advertising path as native tools. See Tools for the complete built-in API and network-safety contract.

Registry & Discovery​

Agent.discover_agents​

async methodprotolink.agents.Agent.discover_agents
source
async discover_agents(
  filter_by: dict[str, Any] | None = None,
) -> list[AgentCard]

Query the configured registry for matching cards. Results are cached per stringified filter only when discovery_ttl is greater than zero.

Parameters

filter_bydict[str, Any] | Nonedefault: None
Optional registry-side card criteria, including nested fields such as {"capabilities.streaming": true}.

Returns

cardslist[AgentCard]
Matching cards, or an empty list when no registry client is configured.

Agent.register / Agent.unregister​

async methodsprotolink.agents.Agent.register / unregister
source
async register() -> None
async unregister() -> None

Register this Agent's current card or remove its URL from the configured registry. Both methods return silently when no registry client exists. Manual register() does not start the automatic heartbeat loop; lifecycle startup owns heartbeat scheduling.

Utility Methods​

Agent.get_agent_card​

methodprotolink.agents.Agent.get_agent_card
source
get_agent_card(*, as_json: bool = True) -> AgentCard | dict[str, Any]

Return the live identity card or its serializable dictionary representation. as_json=True returns a dictionary despite the historical parameter name; it does not return a JSON string.

Parameters

as_jsonbooldefault: True
Keyword-only representation switch. True calls card.to_dict(); false returns the Agent's live AgentCard object rather than a defensive copy.

Agent.get_status / Agent.get_chat​

methodsprotolink.agents.Agent status renderers
source
get_status(
  output_format: Literal["html", "json"] = "html",
) -> str
get_chat() -> str

Render the built-in operational status or chat page. HTML mode and get_chat() return self-contained browser markup. Despite the "json" format name, the current get_status("json") implementation returns str(card.to_dict()), which is a Python dictionary representation rather than guaranteed valid JSON. The chat renderer displays a fallback when no LLM is configured, while POST chat handling requires an LLM and enabled exposure.

Parameters

output_formatLiteral["html", "json"]default: "html"
Format used only by get_status(). HTML renders the operational page; JSON returns the string form of the card dictionary. Any other runtime value raises ValueError. get_chat() takes no arguments.

Agent.handle_chat_message​

async methodprotolink.agents.Agent.handle_chat_message
source
async handle_chat_message(
  data: dict[str, Any],
) -> dict[str, str]

Validate an incoming chat payload, invoke the Agent with its message and session ID, and return a response dictionary. The server route controls whether this handler is exposed.

Parameters

datadict[str, Any]required
Mapping containing message and optionally session_id. A missing or falsey message returns an error mapping; the session defaults to "chat_default". The handler also returns error mappings when no LLM is configured, chat exposure is disabled, or invocation raises.

Agent.llm / Agent.storage​

propertiesprotolink.agents.Agent.llm / storage
source
llm: LLM | None
storage: Storage

The llm setter calls validate_connection() and updates card.capabilities.has_llm from that result. The storage setter updates the existing State object's storage reference so future persistence follows the replacement backend; the annotation expects Storage, but the setter performs no runtime type check.

Existing data
Changing storage does not migrate data from the old backend. Changing llm does not rewrite existing conversation histories or rebuild a running server's route set.

Agent.set_registry​

methodprotolink.agents.Agent.set_registry
source
set_registry(
  registry: TransportType | Registry | RegistryClient | None,
  registry_url: str | None = None,
) -> None

Replace the Agent's discovery client. Passing None disables discovery; a Registry or RegistryClient is adopted; a transport alias constructs a client for registry_url.

Parameters

registryTransportType | Registry | RegistryClient | Nonerequired
Required selection value. A Registry contributes its client, a RegistryClient is retained directly, and a transport alias constructs a new client. A falsey value clears registry_client and logs an error; an unsupported truthy object also clears it.
registry_urlstr | Nonedefault: None
Registry endpoint used only when registry is a transport alias. If omitted for an alias, the method logs an error and returns without replacing the existing registry client.
No automatic registration
Reconfiguration does not register the card or start heartbeats. Call register(), or start the Agent with lifecycle registration enabled.

Agent.sync​

attributeprotolink.agents.Agent.sync
source
sync: SyncAgent

Per-instance blocking facade created during Agent construction. It is an ordinary attribute rather than a class property, and it delegates to this exact Agent instance.

Storage and Persistence​

Protolink provides a storage abstraction to allow agents to persist data across tasks or even standalone.

Core Storage Interface​

The Storage base class defines the CRUD interface:

from protolink.storage import Storage

class MyStorage(Storage):
    def save(self, data): ...
    def load(self): ...
    def update(self, data): ...
    def delete(self): ...

In-Memory Storage (Default)​

Protolink includes a built-in InMemoryStorage which is the default storage backend for all agents. It is a lightweight, RAM-backed dictionary that supports TTL (Time-To-Live) for automatic cleanup.

from protolink.storage import InMemoryStorage

# Default: shared class-level store
storage = InMemoryStorage(namespace="my_agent", ttl=3600)
agent = Agent(card=card, storage=storage)

SQLite Storage​

For persistent storage across restarts, use the built-in SQLiteStorage:

from protolink.storage import SQLiteStorage

storage = SQLiteStorage(db_path="my_agent.db", namespace="main_agent")
agent = Agent(card=card, storage=storage)

State Persistence​

When an agent is initialized with the state parameter, it tracks internal state across multiple task executions based on a session_id.

  1. Activation: Pass a list of state modules to the Agent constructor.

    # Enable conversation history and tool state persistence
    agent = Agent(card=card, state=["conversation", "tools"])

  2. Identification: Include a session_id in your task metadata. This ID is used to partition the state in the storage.

    task = Task.create(Message.user("My name is Alice"))
    task.metadata["session_id"] = "user_123"
    await agent.execute_task(task)

  3. Resumption: Subsequent tasks with the same session_id will automatically load the previous state (e.g., conversation history) into the execution context.

Supported State Modules​

ModuleDescription
conversationPersists LLM conversation history between tasks with the same session_id.
toolsProvides a storage-backed extension point for tool-specific state.
taskProvides a storage-backed extension point for task metadata outside the live Task object.
flowProvides storage-backed flow context; active flow prompts are carried on task.flow_state.
Session IDs

When using direct invocation methods like invoke() or sync.invoke(), a default session_id of "invocation_session_id" is used if none is provided. This ensures that sequential calls to the same agent instance share history by default when state=["conversation"] is enabled.

If no session_id is provided in the task metadata (for non-invoke calls), the agent falls back to using the task.id, effectively making that specific task stateless unless further responses are sent to it.

Chat Gateway​

When an agent is configured with an LLM and uses an HTTP-compatible transport, Protolink automatically exposes a built-in Chat UI at the /chat endpoint. This provides a browser-based interface for interacting with the agent directly, ideal for development, demos, and quick testing.

How It Works​

  • GET /chat - Serves a self-contained HTML/CSS/JS chat interface on HTTP-compatible transports. The endpoint is part of the agent route set and displays a fallback message if no LLM is configured or chat exposure is disabled.
  • POST /chat - Accepts {"message": "...", "session_id": "..."} and returns {"response": "..."}. The endpoint is registered only when the agent has an LLM, and the handler returns an error if chat exposure is disabled.

The chat page displays agent metadata (name, description, skills) and LLM configuration (provider, model, temperature) in a sidebar, alongside a modern conversational interface.

Usage​

No extra setup is needed, just provide an LLM when creating your agent:

from protolink.agents import Agent
from protolink.llms.api import OpenAILLM

agent = Agent(
    card={"name": "assistant", "description": "A helpful assistant", "url": "http://localhost:8000"},
    transport="http",
    llm=OpenAILLM(model="gpt-4o"),
)

agent.start()
# Chat UI is now available at http://localhost:8000/chat
Chat vs Status

The /status page shows the agent's operational health and metadata. The /chat page provides an interactive conversation interface. Both are served automatically by HTTP-compatible agent transports when the agent starts.

YAML Import and Export​

Protolink supports exporting an agent's configuration (identity card, capabilities, transport, TLS file references, LLM, security/authenticator, registered tools, and non-default first-party capability policy) to a YAML file, and importing it back to reconstruct a functional Agent instance. TLS serialization stores certificate paths and settings, never certificate or private-key contents.

Agent serialization methods​

methodsprotolink.agents.Agent serialization
source
to_dict() -> dict[str, Any]
to_yaml_string() -> str
to_yaml(filepath: str) -> None

Agent.from_dict(data: dict[str, Any], **overrides) -> Agent
Agent.from_yaml_string(yaml_str: str, **overrides) -> Agent
Agent.from_yaml(filepath: str, **overrides) -> Agent

Export the reconstructable Agent configuration as Python data or YAML, or create a new Agent from one of those representations. Import methods are class methods: subclasses receive an instance of the subclass.

Parameters

datadict[str, Any]required

Parsed Agent configuration for from_dict().

yaml_strstrrequired

YAML document for from_yaml_string().

filepathstrrequired

Destination for to_yaml() or source for from_yaml().

**overridesAny

Constructor values that replace serialized values during import. Use overrides for environment-specific transports, credentials, executable policies, approval handlers, or other dependencies that should not be trusted or embedded.

Serialization boundary

Configuration export is not a live runtime checkpoint. Active tasks, event loops, open connections, in-memory discovery caches, cancellation tokens, approval callbacks, and arbitrary executable policy objects are not serialized.

Exporting an Agent​

To serialize and export an agent's configuration:

# Export to a YAML file
agent.to_yaml("agent_config.yaml")

# Get configuration as a YAML string
yaml_str = agent.to_yaml_string()

# Get configuration as a dictionary
config_dict = agent.to_dict()

Importing an Agent​

To load and reconstruct an agent from a serialized configuration:

from protolink.agents import Agent

# Reconstruct from a YAML file
agent = Agent.from_yaml("agent_config.yaml")

# Reconstruct from a YAML string
agent = Agent.from_yaml_string(yaml_str)

# Reconstruct from a dictionary
agent = Agent.from_dict(config_dict)

Handling Dependencies and Overrides​

  1. Security & Credentials: Treat exported Agent configuration as sensitive. Configured outbound credentials and authenticator settings can be serialized, including bearer secrets, API-key maps, basic credentials, and OAuth client secrets. Review and protect the file, or pass replacement values during import: 
    agent = Agent.from_yaml("agent_config.yaml", credentials="my-secret-key")
    BRAVE_SEARCH_API_KEY is different: web_search() reads it from the environment only when the default Brave engine is invoked, so the built-in tool does not place that key in Agent dict/YAML output. The explicit engine="duckduckgo" path is keyless.
  2. Built-ins & Policy: Built-in tools serialize by stable first-party identity. A non-default, exact CapabilityPolicy serializes its declarative rules, default effect, and name. Executable custom policies (including CapabilityPolicy subclasses) and approval callbacks are not embedded; pass them as policy= and approval_handler= overrides when importing. An explicit policy override takes precedence over serialized first-party rules.
  3. Tool Function Paths: Standard Python tools are serialized using their module and function name paths (e.g. my_module:my_tool_func). When the agent is imported, Protolink dynamically imports the function. If the module cannot be imported (e.g., if loaded in a different environment), Protolink registers a stub tool that returns a clean runtime error when executed rather than crashing initialization.
  4. MCP Tool Adapters: Model Context Protocol (MCP) tool configs are fully serialized. If the MCP dependencies are installed on the target machine, they will be initialized and bound correctly.

Abstract Methods​

The Agent class provides a default implementation for handle_task that handles tool use and LLM inference automatically. You generally do not need to implement any abstract methods unless you require custom logic.

  • handle_task(task: Task) -> Task: Override this if you need custom task processing logic (e.g., conditional execution, routing).
Minimal Agent Implementation
from protolink.agents import Agent
from protolink.models import AgentCard, Task, Message

class EchoAgent(Agent):
    async def handle_task(self, task: Task) -> Task:
        last = task.get_last_part_content()
        return task.complete(f"Echo: {last}")

Error Handling​

The Agent class includes several error handling patterns:

  • Missing Transport: Construction and start() can operate without a server transport, but outbound call_agent() and send_message_to() raise RuntimeError.
  • Authentication Failures: Returns 401 or 403 responses for invalid auth.
  • Tool Errors: Direct call_tool() calls propagate validation, policy, approval, and tool errors. Task-based tool execution converts ordinary tool failures into an error-bearing tool_output part; policy failures remain raised.
  • Task Processing: Non-streaming engine errors mark the task failed and are re-raised through direct handler calls. The streaming engine emits a TaskErrorEvent and a final failed status event.