Registry

The Registry is Protolink's implementation of the A2A discovery primitive. It gives a multi-agent system a shared address book where each running agent can publish its AgentCard, and where other agents can look up peers by name, role, tags, capabilities, or other card metadata.

Its role is coordination, not orchestration. The registry does not decide which agent should handle a task, route messages between agents, or own workflow state. It answers a narrower A2A question: which agents are currently available, where can they be reached, and what do they say they can do?

That separation keeps agent-to-agent communication explicit. Agents still call each other through their own transports and clients, but they no longer need every peer URL hard-coded at startup. ProtoLink extends the basic A2A registry shape with indexed discovery, liveness heartbeats, optional TTL pruning, persistence hooks, and a browser status view while keeping the core contract centered on AgentCard discovery.

Use a registry when agents should discover each other dynamically instead of hard-coding every peer URL.

Quick Start

from protolink.discovery import Registry

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

# ...start agents that use this registry...

registry.stop()

How It Works

The Registry is both a local runtime object and a transport-backed service. Locally, it stores registered agents by their stable AgentCard.url, keeps liveness metadata in RegistryEntry, and maintains secondary indexes for common discovery filters. Over the transport, RegistryClient sends the same registration, heartbeat, unregistration, and discovery requests that agents use when they connect to a registry URL.

This means there are two useful ways to think about the API:

• Use register(), heartbeat(), unregister(), and discover() when code should talk to a running registry through its configured transport.
• Use count(), list_urls(), get_entry(), and clear() when code owns the in-process Registry object and needs diagnostics, tests, or local administration.

Incoming transport requests are handled by handle_register(), handle_heartbeat(), handle_unregister(), and handle_discover(). Those handler methods update the local store, rebuild indexes as needed, prune expired entries when TTL is enabled, and persist entries when storage is configured. Most application code should use the higher-level client methods or Agent registry integration instead of calling handlers directly.

Browser Status

When the registry runs on an HTTP-compatible transport, it exposes a small browser status page at:

GET <registry-url>/status

The page is served by RegistryServer from the same transport-neutral endpoint table as the JSON registry API. In the default HTTP path, HTTPTransport mounts that endpoint on the Starlette or FastAPI backend and returns HTML generated by protolink.utils.renderers.status.to_registry_status_html(). This keeps the page diagnostic-only: registration and discovery still go through the JSON endpoints such as POST /agents/, GET /agents/, DELETE /agents/, and POST /agents/heartbeat.

Registry statusLive identity, uptime, transport, and discovered-agent count from the running registry.

Agents can receive the live Registry object directly:

from protolink.agents import Agent

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

Or they can connect to a registry by transport type and URL:

from protolink.agents import Agent

agent = Agent(
    card=card,
    transport="http",
    registry="http",
    registry_url="http://localhost:9000",
    registry_heartbeat_interval=15,
)

Filters are optional

Calling discover() with no filters returns all registered agents.

Discovery

cards = await registry.discover({"name": "weather_agent"})
cards = await registry.discover({"role": "worker"})
cards = await registry.discover({"tags": ["weather", "forecast"]})

Discovery returns AgentCard objects. The same filter shape is available through Agent.discover_agents() and RegistryClient.discover().

Transport Integration

The Registry is transport-agnostic. It relies on a Transport implementation to expose its API.

Internally, Registry composes the transport in two directions:

• RegistryClient(transport) is the outbound side. Public methods such as register(), heartbeat(), unregister(), and discover() delegate to this client, which turns each operation into a ClientRequestSpec and calls transport.send(...).
• RegistryServer(self, transport) is the inbound side. start() calls the server, the server builds an EndpointSpec table, and the transport mounts those endpoint specs as real routes.

When you construct a registry with Registry(transport="http", url="http://localhost:9000"), the string transport is resolved through get_transport(...). The resulting transport instance is passed to both the client and the server, so one configured URL defines both where the registry listens and where registry client requests are sent.

The registry endpoint table is:

Operation	Client request	Server handler
Register	POST /agents/ with an AgentCard body	handle_register(card)
Unregister	DELETE /agents/ with {"agent_url": ...}	handle_unregister(agent_url)
Heartbeat	POST /agents/heartbeat with {"agent_url": ...}	handle_heartbeat(agent_url)
Discover	GET /agents/ with query filters	handle_discover(filter_by)
Status page	-	GET /status calls handle_status_html()

For HTTP, HTTPTransport.send() serializes request bodies or query parameters and dispatches them with httpx.AsyncClient. On the server side, HTTPTransport.setup_routes() delegates to the selected ASGI backend, currently Starlette or FastAPI, which turns each EndpointSpec into a concrete route. HTTPTransport.start() then starts the ASGI server at the transport URL.

Agents use the same path. If an Agent receives a live Registry object, it extracts registry.client. If it receives a transport string plus registry_url, it creates a separate transport and wraps it in RegistryClient. During agent startup, the agent registers its card through that client; if registry_heartbeat_interval is configured, it keeps sending heartbeats through the same client; and Agent.discover_agents() calls RegistryClient.discover().

Currently supported by the default runtime path:

• HTTPTransport via transport="http"

The transport is responsible for:

• Binding to a host and port
• Exposing registry endpoints
• Handling request/response lifecycle

Discovery moduleRegistry

The discovery service for registering agent cards, finding peers by metadata, maintaining secondary indexes, and exposing registry operations through a transport.

protolink.discovery.Registry

Agent registrationFiltered discoveryRole and tag indexesTTL heartbeatsOptional persistence

RegisterStore an AgentCard and make the agent discoverable through local or remote registry APIs.register()

DiscoverFind peers by name, role, tags, capabilities, or other supported card metadata.discover()

LifecycleRun a registry server in blocking mode or an isolated background thread.start()

LivenessUse TTL and heartbeat settings to prune stale agents in longer-running systems.entry_ttl_seconds

Lifecycle Methods

These methods control the registry server component lifecycle.

Name	Parameters	Returns	Description
start()	background: bool = False	None	Starts the Registry runtime. Can run in the main loop or as an isolated background thread.
stop()	-	None	Stops the Registry runtime and synchronously cleans up resources.

Execution Models

• background=True starts the registry in a dedicated background thread with its own isolated asyncio event loop and returns immediately. Use this for examples, notebooks, tests, and multi-agent scripts.
• background=False blocks the main thread until the registry is stopped. Use this for a standalone registry process.

Common Usage Patterns

Standalone registry service

from protolink.discovery import Registry

registry = Registry(url="http://localhost:9000", transport="http")
registry.start()

Multi-agent orchestration

registry.start(background=True)
agent.start(background=True)

# ...run your orchestration...

agent.stop()
registry.stop()

Graceful shutdown

Always use registry.stop() to cleanly shut down the server and release ports. In blocking scripts, registry.start(background=False) handles KeyboardInterrupt automatically.

Discovery Performance

The Registry is optimized for high-throughput environments where many agents may be registered simultaneously.

Secondary Indexing

To avoid linear scans during common discovery queries, the Registry maintains secondary indexes for:

• Agent name
• Agent role
• Tags

When multiple indexed filters are applied, the Registry performs set intersections and then refines matches for any non-indexed fields.

Robust Fallback

If a query uses non-indexed fields, or if the indexed path returns no candidates while agents are still present, the Registry falls back to a full scan to preserve correctness.

Liveness And Persistence

By default, the registry is an in-memory discovery service. For longer-running systems, two optional knobs make it more robust:

• entry_ttl_seconds prunes agents whose last_seen timestamp is older than the configured TTL.
• storage persists registered entries through the generic Storage interface, so a registry can rebuild its in-memory indexes after restart.

Agents can keep their entry fresh by setting registry_heartbeat_interval on the Agent. After successful registration, the agent periodically calls RegistryClient.heartbeat(agent_url). Heartbeats update liveness metadata only; they do not mutate the agent card or discovery indexes.

from protolink import Agent, AgentCard
from protolink.discovery import Registry
from protolink.storage import SQLiteStorage

registry = Registry(
    url="http://localhost:9000",
    transport="http",
    entry_ttl_seconds=45,
    storage=SQLiteStorage("registry.db", namespace="registry"),
)

agent = Agent(
    AgentCard(name="worker", description="Worker", url="http://localhost:9010"),
    transport="http",
    registry="http",
    registry_url="http://localhost:9000",
    registry_heartbeat_interval=15,
)

The user-facing Registry surface in protolink.discovery.registry.Registry includes:

Method	Purpose
register(card)	Send an agent card to the registry transport for registration.
heartbeat(agent_url)	Refresh last_seen for a registered agent through the registry transport.
unregister(agent_url)	Remove an agent through the registry transport.
discover(filter_by=None)	Return matching live AgentCard objects through the registry transport.
list_urls()	Return registered agent URLs from the local in-process registry store.
count()	Return the number of live local entries after TTL pruning.
clear()	Clear local entries, secondary indexes, and persisted registry state if storage is configured.
get_entry(agent_url)	Return local RegistryEntry liveness metadata for diagnostics and tests.

The handle_* methods are the server-side endpoint hooks used by RegistryServer. They are part of the served registry implementation, but most callers should not need them unless they are writing tests, custom transports, or an alternate registry server.

Constructor

Registry(
    transport: TransportType | Transport = "http",
    url: str | None = None,
    verbosity: Literal[0, 1, 2] = 1,
    *,
    entry_ttl_seconds: float | None = None,
    storage: Storage | None = None,
)

Parameter	Type	Default	Description
transport	`TransportType	Transport`	"http"
url	`str	None`	None
verbosity	Literal[0, 1, 2]	1	Logging verbosity: 0 = warning, 1 = info, 2 = debug.
entry_ttl_seconds	`float	None`	None
storage	`Storage	None`	None

Single source of truth

The Registry's public URL is derived from its transport and used by agents for registration and discovery.

URL Handling

Both Agents and the Registry expose a url property through their transport.

from protolink.transport import HTTPTransport

transport = HTTPTransport(url="http://localhost:9000")
registry = Registry(transport=transport)

This keeps host, port, transport, and discovery metadata consistent across agent and registry instances.