Skip to main content

Server

Servers in Protolink act as the coordination layer between business logic (Agents or Registries) and the underlying Transport mechanism. They are responsible for wiring endpoints, managing lifecycle, and ensuring that the core logic remains transport-agnostic.

AgentServer always binds ProtoLink's native endpoints. When an HTTP agent is created with a2a=True, it additionally binds the A2A 1.0 adapter to the same execution logic. Agent authors still implement handle_task(Task) once.

Server coordination layerAgentServer and RegistryServer

The endpoint declaration layer that binds Agent and Registry behavior to whatever transport backend is running the process.

protolink.server
EndpointSpec-drivenTransport-agnosticAgent and registry serversStreaming routesA2A 1.0 HTTP routesStatus and chat endpoints
AgentServerExposes native task, streaming, discovery, cancellation, status, chat, and LLM control endpoints, with an opt-in A2A 1.0 JSON-RPC boundary./tasks/
RegistryServerExposes registration, deregistration, discovery, liveness, and status endpoints./agents/
EndpointSpecKeeps route definitions declarative so transports can bind the same contract to different backends.EndpointSpec
LifecycleStarts and stops the transport without moving business behavior into network adapters.start() / stop()

Concept​

A Server does not implement networking itself. Instead, it:

  1. Defines Endpoints: Declares the API surface (paths, methods, handlers).
  2. Binds Handlers: Connects these endpoints to the implementation (Agent or Registry).
  3. Manages Lifecycle: Starts and stops the underlying transport.

This separation allows an Agent to run over HTTP, WebSocket, or in-memory transports without changing a single line of agent code.

AgentServer​

The AgentServer exposes an Agent over a Transport.

Responsibilities​

  • Exposing the Task submission endpoint.
  • Exposing the Task streaming endpoint when the transport supports streaming.
  • Serving the Agent's identity card (/.well-known/agent.json).
  • On HTTP with a2a=True, serving the A2A 1.0 Agent Card and JSON-RPC adapter.
  • Providing a status page.
  • Exposing the Chat Gateway when the agent has an LLM.

Endpoints​

EndpointMethodDescription
/tasks/POSTTask Submission. Accepts a Task object, processes it via the Agent, and returns the result.
/tasks/cancelPOSTTask Cancellation. Parses a TaskCancellationRequest and asks the Agent to cancel active work.
/tasks/streamPOSTTask Streaming. Accepts a Task object and streams status, LLM, tool, artifact, and final events. Only registered when the transport has supports_streaming=True.
/llm/history/compactPOSTHistory Compaction. Runs the Agent's explicit conversation-compaction control plane.
/state/describePOSTState Description. Reports enabled persistent state stores.
/state/resetPOSTState Reset. Clears selected persistent state stores after policy authorization.
/state/compactPOSTState Compaction. Compacts selected persistent state stores after policy authorization.
/.well-known/agent.jsonGETAgent Discovery. Returns the AgentCard describing this agent.
/.well-known/agent-card.jsonGETA2A 1.0 Agent Card. Added only by Agent(..., transport="http", a2a=True); returns the standard wire card.
/POSTA2A 1.0 JSON-RPC. Added only with a2a=True; handles SendMessage, GetTask, ListTasks, and CancelTask.
/statusGETStatus Page. Returns a human-readable HTML status dashboard.
/healthzGETHealth. Returns the underlying transport's health snapshot.
/readyzGETReadiness. Currently calls the same transport health method as /healthz.
/chatGETChat Page. Returns a self-contained HTML/CSS/JS chat interface. Always registered; shows a fallback message if no LLM is configured.
/chatPOSTChat Message. Accepts {"message": "...", "session_id": "..."} and returns the agent's response. Only registered when the agent has an LLM.

The two A2A 1.0 routes require a transport whose transport_type is "http" and a2a=True. With the default False, HTTP retains the native endpoint contract and does not expose /.well-known/agent-card.json or POST /. Other transports are not presented as A2A 1.0 wire bindings. See A2A Core and 1.0 Compatibility for the implemented scope and TCK evidence.

Usage​

The Agent class automatically creates an AgentServer internally when a transport is provided. You rarely need to instantiate AgentServer or wire the A2A adapter directly.

# AgentServer is created internally; A2A routes are explicit and additive.
agent = Agent(card=card, transport="http", a2a=True)

agent.start()

RegistryServer​

The RegistryServer exposes a Registry over a Transport.

Responsibilities​

  • Handling agent registration and deregistration.
  • Serving the discovery endpoint for finding agents.
  • Providing a status page.

Endpoints​

EndpointMethodDescription
/agents/POSTRegister. Registers an agent with the registry. Body: AgentCard.
/agents/DELETEUnregister. Removes an agent. Body: {"agent_url": "..."}.
/agents/heartbeatPOSTHeartbeat. Refreshes liveness for one agent URL.
/agents/GETDiscover. Returns agents matching direct query filters such as ?name=worker&role=worker.
/statusGETStatus Page. Returns a human-readable HTML status dashboard.
/healthzGETHealth. Returns the underlying transport's health snapshot.
/readyzGETReadiness. Currently calls the same transport health method as /healthz.

Usage​

from protolink.discovery.registry import Registry
from protolink.transport import HTTPTransport
from protolink.server.registry import RegistryServer

# Create logic and transport
transport = HTTPTransport(url="http://localhost:8000")
registry = Registry(transport=transport)

# Create Server (wiring)
server = RegistryServer(registry, transport)

# Start
await server.start()

Architecture​

The server architecture relies on the EndpointSpec model to define routes in a generic way.

EndpointSpec​

The EndpointSpec class (defined in protolink.server.endpoint_handler) is the contract between a Server and a Transport.

@dataclass(frozen=True)
class EndpointSpec:
    name: str              # Internal unique name for the endpoint
    path: str              # URL path (e.g. "/tasks/")
    method: HttpMethod     # HTTP Method (GET, POST, etc.)
    handler: Callable      # Async function to process the request
    
    # Configuration
    content_type: Literal["json", "html"] = "json"
    streaming: bool = False
    mode: Literal["request_response", "stream"] = "request_response"
    
    # Request Parsing
    request_parser: Callable[[Any], Any] | None = None
    request_source: RequestSourceType = "none"

How it Works​

  1. Transport-Agnostic Definition: The Server creates a list of EndpointSpec objects describing what it needs to expose.
  2. Transport Implementation: The Transport iterates over these specs and registers them with its underlying web framework (e.g., Starlette or FastAPI).
  3. Request Handling:
    • The Transport receives a raw HTTP request.
    • It extracts data based on request_source (e.g., reads the body JSON).
    • It passes this data to the request_parser (if defined) to convert it into a domain object.
    • It calls the handler with the domain object.
    • For request/response endpoints, it serializes the result back to the wire format.
    • For streaming endpoints, it iterates the handler and serializes each event until a final event closes the stream.

This design ensures that your Agent logic deals only with Task and Message objects, never raw HTTP requests, while the Transport handles the nitty-gritty of networking protocols.

Server API reference​

The two server classes are the public package surface. EndpointSpec and EndpointRequest live in protolink.server.endpoint_handler; they are included here because custom transports and protocol adapters need the same declarative route contract even though those types are not re-exported from protolink.server.

AgentInterface server protocol​

protocolprotolink.server.agent.AgentInterface
source
class AgentInterface(Protocol)

Describe the structural surface that AgentServer calls. An object does not need to inherit from this protocol; it only needs to provide compatible attributes and methods.

Required surface

cardAgentCard

Public identity and capabilities.

handle_task / run_taskasync (Task) -> Task

Business handler and its cancellation-aware execution wrapper.

handle_task_streaming / run_task_streaming(Task) -> AsyncIterator[Any]

Business event stream and its cancellation-aware wrapper.

cancel_taskasync (TaskCancellationRequest) -> Task

Active-task cancellation control.

compact_historyasync (HistoryCompactionRequest) -> HistoryCompactionResult

Explicit LLM history compaction.

describe_state / reset_state / compact_stateasync (StateOperationRequest) -> StateOperationResult

Persistent-state control-plane handlers.

get_agent_card(*, as_json: bool = True) -> AgentCard | dict[str, Any]

Native discovery document.

get_status(output_format: "html" | "json" = "html") -> str

Render Agent status as HTML or, for "json", the current Python str(card.to_dict()) representation. That branch is not guaranteed to be valid JSON.

get_chat() -> str

Render the HTML chat page.

handle_chat_messageasync (dict[str, Any]) -> dict[str, str]

Chat message endpoint handler.

Name collision

This internal server protocol is not re-exported from protolink.server. The public protolink.AgentInterface name refers instead to the Agent Card interface dataclass containing a URL, transport, and protocol version.

AgentServer​

classprotolink.server.AgentServer
source
class AgentServer(
  transport: Transport,
  agent: AgentInterface,
  *,
  a2a: bool = False,
)

Bind an Agent-compatible object to a configured transport. Construction records the collaborators and validates A2A compatibility; endpoint registration and network startup are deferred until start().

Parameters

transportTransportrequired

Concrete transport that receives endpoint specifications and owns the listening lifecycle. Passing None is rejected immediately.

agentAgentInterfacerequired

Structurally compatible object implementing task execution, streaming, cancellation, history/state controls, card/status/chat rendering, and chat message handling. Runtime inheritance from the protocol is not required.

a2abooldefault: False

Build an A2A 1.0 JSON-RPC adapter and its two additional endpoints. When enabled, transport.transport_type must equal "http".

Raises

ValueError

Raised when no transport is supplied, or when a2a=True is paired with a non-HTTP transport type.

adapter construction error

A2A adapter initialization errors propagate when A2A support is enabled.

Endpoint timing

Constructing the server does not call setup_routes(). Route declarations are built during the first successful start().

AgentServer.start​

async methodprotolink.server.AgentServer.start
source
await start() -> None

Build the complete endpoint table, pass it to the transport, and await the transport's server startup.

Returns

NoneNone

The server records itself as running only after transport.start() completes successfully.

Side effects

native routesEndpointSpec[]

Registers task submission/cancellation, history and state controls, card, status, health/readiness, and chat-page endpoints.

task streamconditional route

Adds POST /tasks/stream only when the transport advertises supports_streaming=True.

chat messageconditional route

Adds POST /chat when the Agent has an llm, or its card capabilities advertise has_llm=True.

A2A routesconditional routes

Adds the standard Agent Card and root JSON-RPC routes when the server owns an A2A adapter.

Raises

route or transport error

Exceptions from endpoint construction, setup_routes(), or transport.start() propagate. The running flag remains False if startup does not complete.

Idempotence

Calling start() again while the server is running is a no-op. A stop-then-start cycle builds and submits the routes again; whether duplicate route registration is accepted depends on the transport implementation.

AgentServer.stop​

async methodprotolink.server.AgentServer.stop
source
await stop() -> None

Stop the underlying transport and close the optional A2A adapter.

Returns

NoneNone

Returns immediately when the server is not marked as running.

Shutdown order

1transport

Await transport.stop().

2A2A adapter

Await adapter closure when A2A support was enabled.

3server state

Mark the server idle after both preceding operations complete.

Raises

shutdown error

Transport and A2A closure exceptions propagate. If either operation fails, the running flag is not cleared by the current implementation.

RegistryInterface server protocol​

protocolprotolink.server.registry.RegistryInterface
source
class RegistryInterface(Protocol)

Describe the structural registry behavior consumed by RegistryServer.

Required surface

handle_registerasync (AgentCard) -> dict[str, str]

Store or replace one agent card.

handle_unregisterasync (agent_url: str) -> dict[str, str]

Remove one URL.

handle_heartbeatasync (agent_url: str) -> dict[str, str]

Refresh liveness metadata.

handle_discoverasync (filter_by: dict[str, Any] | None = None) -> list[dict[str, Any]] | list[AgentCard]

Return cards matching optional filters.

handle_status_html() -> str

Render the registry status page.

Structural typing

The protocol is a server-module implementation detail and is not re-exported. Ordinary applications pass a Registry instance rather than implementing it directly.

RegistryServer​

classprotolink.server.RegistryServer
source
class RegistryServer(
  registry: RegistryInterface,
  transport: Transport,
)

Bind a Registry-compatible object to its transport endpoint table. As with AgentServer, the registry object supplies business behavior and the transport supplies route binding and networking.

Parameters

registryRegistryInterfacerequired

Structurally compatible object implementing register, unregister, heartbeat, discovery, and HTML status handlers.

transportTransportrequired

Concrete route-binding and server-lifecycle implementation.

Raises

ValueError

Raised when transport is None. The registry argument is stored without runtime validation.

RegistryServer.register_parser​

async methodprotolink.server.RegistryServer.register_parser
source
await register_parser(
  request: Any,
) -> AgentCard

Convert an inbound registration body into ProtoLink's runtime AgentCard.

Parameters

requestAnyrequired

Value forwarded directly to AgentCard.from_dict(). Normal transport usage supplies a decoded mapping.

Returns

cardAgentCard

Normalized agent identity and capability model passed to the registry's registration handler.

Raises

deserialization error

Mapping-shape, required-field, and nested model errors from AgentCard.from_dict() propagate.

RegistryServer.unregister_parser​

async methodprotolink.server.RegistryServer.unregister_parser
source
await unregister_parser(
  request: Any,
) -> str

Read the agent_url value from an unregistration request body.

Parameters

requestAnyrequired

Mapping-like object expected to provide .get("agent_url").

Returns

agent_urlstr

URL forwarded to handle_unregister(). Despite the annotation, a missing key currently produces None.

Raises

AttributeError

Raised when the request has no compatible .get() method.

RegistryServer.heartbeat_parser​

async methodprotolink.server.RegistryServer.heartbeat_parser
source
await heartbeat_parser(
  request: Any,
) -> str

Read the agent_url value from a heartbeat body.

Parameters

requestAnyrequired

Mapping-like object expected to provide .get("agent_url").

Returns

agent_urlstr

URL forwarded to handle_heartbeat(). A missing key currently yields None despite the declared return type.

Raises

AttributeError

Raised when the request has no compatible .get() method.

RegistryServer.discover_parser​

async methodprotolink.server.RegistryServer.discover_parser
source
await discover_parser(
  request: Any,
) -> dict[str, Any] | None

Normalize discovery query data into the filter mapping expected by the registry handler.

Parameters

requestAnyrequired

Decoded query mapping. A wrapper key named filter_by is unwrapped when present.

Returns

filter_bydict[str, Any] | None

None for a non-dictionary request; the nested filter_by value when that key exists; otherwise the request dictionary itself.

Current validation

A nested filter_by value is returned without checking that it is a dictionary. Normal RegistryClient discovery sends filter keys directly, such as ?name=worker; a raw ?filter_by={...} query value is only a string and is not JSON-decoded by this parser.

RegistryServer.start​

async methodprotolink.server.RegistryServer.start
source
await start() -> None

Register all registry, status, health, and readiness routes, then await transport startup.

Returns

NoneNone

The running flag is set only after the transport starts successfully. Repeated calls while running return immediately.

Raises

route or transport error

Exceptions from setup_routes() or transport.start() propagate.

RegistryServer.stop​

async methodprotolink.server.RegistryServer.stop
source
await stop() -> None

Await transport shutdown and mark the registry server idle.

Returns

NoneNone

Returns immediately when the server is not running.

Raises

transport error

Shutdown exceptions propagate, and the running flag remains set when transport shutdown fails.

Endpoint declaration reference​

EndpointSpec​

frozen dataclassprotolink.server.endpoint_handler.EndpointSpec
source
class EndpointSpec(
  name: str,
  path: str,
  method: HttpMethod,
  handler: Callable[..., Any],
  content_type: Literal["json", "html"] = "json",
  streaming: bool = False,
  mode: Literal["request_response", "stream"] = "request_response",
  request_parser: Callable[[Any], Any] | None = None,
  request_source: RequestSourceType = "none",
)

Describe one transport-neutral route. Server classes create these immutable declarations; each transport interprets their path, extraction, parsing, invocation, serialization, and streaming fields for its own backend.

Parameters

namestrrequired

Logical route identifier. EndpointSpec itself does not enforce uniqueness.

pathstrrequired

Protocol-facing route path, conventionally beginning with /. No syntax validation is performed by the dataclass.

method"GET" | "POST" | "DELETE" | "PUT" | "PATCH"required

HTTP-style operation used by applicable transports.

handlerCallable[..., Any]required

Synchronous function, coroutine function, or streaming callable invoked after extraction and optional parsing.

content_type"json" | "html"default: "json"

Response rendering mode interpreted by the HTTP/ASGI backends. WebSocket, gRPC, and Runtime transports serialize returned string values through their normal protocol envelope instead.

streamingbooldefault: False

Signals that the handler returns an asynchronous stream rather than one response value.

mode"request_response" | "stream"default: "request_response"

Explicit execution mode consumed by transports. It is separate from streaming; the dataclass does not require the two fields to agree.

request_parserCallable[[Any], Any] | Nonedefault: None

Optional transformer applied to the extracted request value before handler invocation. Transports may support synchronous and asynchronous parser results.

request_sourceRequestSourceTypedefault: "none"

Select none, body, query_params, form, headers, path_params, or request. The last form supplies an EndpointRequest containing several request facets.

Frozen, not validated

Attribute reassignment is blocked by the frozen dataclass, but contained callables and other referenced values remain mutable. Construction does not validate paths, method strings at runtime, handler callability, unique names, or streaming-field consistency.

Request-source support

The shared type includes form, but current built-in transports do not extract form data. HTTP backends support body, query, headers, path parameters, and the combined request view; WebSocket, gRPC, and Runtime currently bind body and query-parameter sources.

EndpointRequest​

frozen dataclassprotolink.server.endpoint_handler.EndpointRequest
source
class EndpointRequest(
  body: Any = None,
  query_params: Mapping[str, str] = field(default_factory=dict),
  path_params: Mapping[str, str] = field(default_factory=dict),
  headers: Mapping[str, str] = field(default_factory=dict),
  method: str = "",
  url: str = "",
  principal_id: str | None = None,
)

Provide a small, framework-independent view of an inbound HTTP-style request. It is used by endpoints whose adapters need more than one extracted source, such as A2A JSON-RPC requests that also need authenticated-principal information.

Parameters

bodyAnydefault: None

Decoded or raw body value supplied by the transport.

query_paramsMapping[str, str]default: {}

Query parameter view. A new dictionary is used when omitted.

path_paramsMapping[str, str]default: {}

Route parameter view. A new dictionary is used when omitted.

headersMapping[str, str]default: {}

Request header view. Header normalization depends on the transport.

methodstrdefault: ""

Incoming protocol method.

urlstrdefault: ""

Incoming request URL as supplied by the transport.

principal_idstr | Nonedefault: None

Authenticated principal propagated by the transport when available.

Shallow immutability

The dataclass is frozen and slot-based, but mappings supplied by the caller are not copied or wrapped. Their contents can still change after the request object is constructed.