Registry

The Registry is ProtoLink's first-class discovery service for A2A-based agent meshes. It uses AgentCard, ProtoLink's runtime form of A2A's identity and capability primitive, as a shared address book where running agents publish who they are and other agents find peers by name, role, tags, capabilities, endpoint, 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 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. A2A 1.0 standardizes per-agent card discovery; ProtoLink adds central indexed lookup, roles and tags, liveness heartbeats, optional TTL pruning, persistence hooks, and a browser status view around that model. Those registry endpoints are ProtoLink-native runtime services, not additional A2A 1.0 operations.

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()

View the registry in Devtools

While the registry is running, open it with protolink dashboard --registry-url http://localhost:9000 --open.

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.

Registry.start

methodprotolink.discovery.Registry.start

source↗

start(
  *,
  background: bool = False,
) -> None

Start the RegistryServer and keep its transport lifecycle alive. This public entry point is synchronous even though server startup and shutdown are asynchronous.

Parameters

backgroundbooldefault: False

When false, run the lifecycle with asyncio.run() and block the caller. When true, start a non-daemon thread with its own event loop, wait for readiness, and return.

Returns

NoneNone

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

Raises

startup error

Transport binding, route setup, and server failures propagate. Background failures are captured in the lifecycle thread and re-raised to the caller.

Active event loops

Blocking mode logs an error but still calls asyncio.run(); from an active loop that raises RuntimeError. Use background=True in async applications and notebooks.

Registry.stop

methodprotolink.discovery.Registry.stop

source↗

stop() -> None

Cancel the private background lifecycle task and synchronously wait up to ten seconds for its thread to exit. Lifecycle cancellation calls the server's async stop path, which closes the shared transport.

Returns

NoneNone

Returns after the background thread exits or the join timeout elapses.

Repeated calls

The internal async stop primitive is guarded, and calling the public method again is safe.

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:

Registry.register

async methodprotolink.discovery.Registry.register

source↗

async register(
  card: AgentCard,
) -> dict[str, str]

Send an AgentCard to this Registry's transport-backed client. This does not update the local store directly; the served request returns through handle_register(), even when client and server belong to the same Registry object.

Parameters

cardAgentCardrequired

Complete identity and capability card keyed by its stable URL on the server.

Returns

statusdict[str, str]

Server status payload. Unlike RegistryClient, this facade catches any exception, logs it, and returns {"status": str(error)}.

Retry contract

The built-in register request is not idempotent and is not automatically retried by transport policy.

Registry.heartbeat

async methodprotolink.discovery.Registry.heartbeat

source↗

async heartbeat(
  agent_url: str,
) -> dict[str, str]

Ask the running registry service to refresh one entry's liveness timestamp.

Parameters

agent_urlstrrequired

Stable URL used as the entry key.

Returns

statusdict[str, str]

Success, not-found, or caught-error status. A heartbeat never changes the card or secondary indexes.

Registry.unregister

async methodprotolink.discovery.Registry.unregister

source↗

async unregister(
  agent_url: str,
) -> dict[str, str]

Send an idempotent removal request through the configured RegistryClient.

Parameters

agent_urlstrrequired

Stable URL to remove.

Returns

statusdict[str, str]

Server or caught-error status. Removing an unknown URL still returns the handler's success message.

Registry.discover

async methodprotolink.discovery.Registry.discover

source↗

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

Query the running service through RegistryClient and reconstruct matching AgentCard objects.

Parameters

filter_bydict[str, Any] | Nonedefault: None

Exact field filters. Name, role, and a single string tag use secondary indexes; other fields and tag lists are refined by the full matcher.

Returns

cardslist[AgentCard]

All live cards when no filter is supplied, otherwise exact matches after TTL pruning.

Raises

transport, decoding, or model error

Unlike the three status-returning facade methods, discover() does not catch RegistryClient failures.

Registry.list_urls / Registry.count

methodsprotolink.discovery.Registry local inspection

source↗

list_urls() -> list[str]
count() -> int

Inspect the in-process store without a transport request. Both methods first prune expired entries and persist the pruned store when storage is configured. list_urls() preserves dictionary insertion order; count() returns the number of currently live entries.

Registry.get_entry

methodprotolink.discovery.Registry.get_entry

source↗

get_entry(
  agent_url: str,
) -> RegistryEntry | None

Return local liveness metadata for one URL after TTL pruning.

Parameters

agent_urlstrrequired

Local entry key.

Returns

entryRegistryEntry | None

The stored entry object or None. This is the live object, not a defensive copy.

Registry.clear

methodprotolink.discovery.Registry.clear

source↗

clear() -> None

Remove every local card, RegistryEntry, and secondary-index value, then persist the empty entry list when storage is configured.

Administrative mutation

This bypasses transport, authentication, and server handlers. It is intended for code that owns the in-process Registry, tests, and explicit administration.

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.

Registry.handle_register

async handlerprotolink.discovery.Registry.handle_register

source↗

async handle_register(
  card: AgentCard,
) -> dict[str, str]

Insert or replace a local registration. Replacement first runs unregistration cleanup, then writes a fresh RegistryEntry with last_seen=time.time(), updates name/role/tag indexes, and persists the full entry set.

Parameters

cardAgentCardrequired

Validated card supplied by RegistryServer's request parser or a direct caller.

Returns

statusdict[str, str]

Always {"status": "agent registered successfully"} after mutation; storage errors propagate.

Registry.handle_heartbeat

async handlerprotolink.discovery.Registry.handle_heartbeat

source↗

async handle_heartbeat(
  agent_url: str,
) -> dict[str, str]

Prune stale entries, then replace one RegistryEntry with a fresh timestamp while retaining its card and metadata.

Parameters

agent_urlstrrequired

Stable local entry key.

Returns

statusdict[str, str]

"agent heartbeat recorded" or "agent not found". The not-found result is a normal payload, not an exception.

Registry.handle_unregister

async handlerprotolink.discovery.Registry.handle_unregister

source↗

async handle_unregister(
  agent_url: str,
) -> dict[str, str]

Remove one local card and entry, clean empty secondary-index buckets, and persist the resulting store. The operation is idempotent.

Parameters

agent_urlstrrequired

Stable entry key to remove.

Returns

statusdict[str, str]

Success status even when the URL was already absent.

Registry.handle_discover

async handlerprotolink.discovery.Registry.handle_discover

source↗

async handle_discover(
  filter_by: dict[str, Any] | None = None,
  *,
  as_json: bool = False,
) -> list[dict[str, Any]] | list[AgentCard]

Prune expired registrations, select indexed candidates, refine every candidate with exact field matching, and optionally serialize results.

Parameters

filter_bydict[str, Any] | Nonedefault: None

Exact AgentCard attributes. For tags, every requested tag must be present.

as_jsonbooldefault: False

Return dictionaries for transport handlers or AgentCard objects for local callers.

Returns

cardslist[dict[str, Any]] | list[AgentCard]

Live matches in registry insertion order or candidate-set iteration order, depending on the filter path.

Fallback scan

When indexes yield no candidates while cards exist, the handler scans all cards. This preserves correctness when tests or compatibility code manipulate the public-ish _agents dictionary without rebuilding indexes.

Registry.handle_status_html

handlerprotolink.discovery.Registry.handle_status_html

source↗

handle_status_html() -> str

Prune expired entries and render the current registry card set and uptime as a self-contained HTML status page.

Returns

htmlstr

Complete diagnostic page markup.

Current renderer label

The current implementation passes the literal transport label "HTTP" to the renderer rather than reading the configured transport type.

Constructor

Registry

classprotolink.discovery.Registry

source↗

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,
)

Create an in-process indexed registry, a RegistryClient, and a RegistryServer around one resolved transport. Construction restores persisted entries and rebuilds secondary indexes, but it does not start listening.

Parameters

transportTransportType | Transportdefault: "http"

Registered transport alias or configured instance. An alias is resolved with url; a concrete transport is shared unchanged by the client and server.

urlstr | Nonedefault: None

Address required when transport is a string. A concrete transport owns its own URL and ignores this argument.

verbosityLiteral[0, 1, 2]default: 1

Registry logger level: warning, info, or debug.

entry_ttl_secondsfloat | Nonedefault: None

Maximum age since last_seen. Expiry is lazy: pruning runs during discovery, status rendering, inspection, heartbeat, and persisted-state load rather than on a timer.

storageStorage | Nonedefault: None

Optional persistence for a single dictionary containing serialized entries. Every registration, heartbeat, removal, clear, or TTL-prune rewrites that payload through storage.save().

Attributes

clientRegistryClient

Read-only outbound facade using the shared transport.

start_timefloat | None

Unix timestamp set after successful server start.

Raises

ValueError

A transport alias lacks url, or transport is neither a registered string nor a Transport instance.

storage/model error

Malformed persisted entries, storage-load failures, and reconstruction errors propagate during construction.

Persisted TTL

Restored entries retain their original last_seen timestamps and are pruned immediately after loading when a TTL is configured.

Registry follows the same construction rule as Agent and AgentClient. A string alias creates a default transport for fast setup; a concrete transport carries TLS, limits, retries, keepalive, and protocol-specific settings. The Registry passes that exact instance to both RegistryClient and RegistryServer, so inbound serving and outbound registry calls share one capability, health, and metrics surface.

# Simple: defaults are sufficient
registry = Registry(
    transport="http",
    url="http://127.0.0.1:9000",
)

# Advanced: configure the service boundary explicitly
from protolink import RetryPolicy, TLSConfig, TransportConfig, TransportLimits
from protolink.transport import HTTPTransport

transport = HTTPTransport(
    url="https://registry.internal:9000",
    tls=TLSConfig(
        certfile="certs/registry.pem",
        keyfile="certs/registry-key.pem",
        cafile="certs/ca.pem",
    ),
    config=TransportConfig(
        limits=TransportLimits(max_concurrent_requests=300),
        retry=RetryPolicy(max_attempts=3),
    ),
)
registry = Registry(transport=transport)

The Registry needs the same protections as an Agent even though its requests are smaller. In a large deployment, many Agents may start or heartbeat at once. Concurrency limits keep that burst bounded, payload limits prevent malformed cards from consuming excessive memory, and health metrics reveal whether discovery traffic is failing or saturating the service. Keeping those settings on its transport also allows the Registry to use a different certificate identity and capacity policy from every Agent that calls it.

The built-in registry request specs mark unregister, heartbeat, and discover as idempotent. register is intentionally not retried automatically because registration may have application-specific replacement semantics. See ClientRequestSpec and the retry contract.

In simple terms, reading discovery results, refreshing the same heartbeat, or removing an already removed URL has a repeatable outcome. Registration can mean “create,” “replace,” or trigger custom persistence behavior, so ProtoLink does not assume that repeating it is harmless. Applications that provide durable idempotent registration semantics can define an explicit custom request contract.

Single source of truth

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

RegistryEntry API

RegistryEntry is the persisted liveness envelope around one AgentCard. Registry users usually obtain it through registry.get_entry().

RegistryEntry

dataclassprotolink.core.registry.RegistryEntry

source↗

RegistryEntry(
  card: AgentCard,
  last_seen: float,
  metadata: dict[str, Any] = field(default_factory=dict),
)

Pair a registered card with its last successful registration or heartbeat time and optional registry-owned metadata.

Parameters

cardAgentCardrequired

Registered identity and capabilities.

last_seenfloatrequired

Unix timestamp used for TTL comparison.

metadatadict[str, Any]default: {}

Per-entry metadata created through a dataclass default factory, so entries do not share one dictionary.

RegistryEntry.is_expired

methodprotolink.core.registry.RegistryEntry.is_expired

source↗

is_expired(
  ttl_seconds: float | None,
  *,
  now: float,
) -> bool

Compare an explicit clock value with last_seen.

Parameters

ttl_secondsfloat | Nonerequired

None disables expiration. Zero or negative values make any entry with a positive age expire; the implementation does not validate positivity.

nowfloatrequired

Caller-supplied Unix time, which makes expiry tests deterministic.

Returns

expiredbool

True only when a TTL exists and now - last_seen > ttl_seconds. Equality is still live.

RegistryEntry.to_dict / RegistryEntry.from_dict

methodsprotolink.core.registry.RegistryEntry serialization

source↗

to_dict() -> dict[str, Any]
RegistryEntry.from_dict(data: dict[str, Any]) -> RegistryEntry

Serialize an entry into card, timestamp, and metadata fields or reconstruct it from that representation. from_dict() requires a "card" mapping, converts last_seen to float with a zero default, and copies metadata into a new dictionary.

Raises

KeyError | TypeError | ValueError

Missing or malformed card data, a non-numeric timestamp, or invalid AgentCard fields.

RegistryClient API

RegistryClient is the thin transport-facing contract used by Registry and Agent. It does not own local entries or indexes and does not catch transport failures.

RegistryClient

classprotolink.client.RegistryClient

source↗

RegistryClient(
  transport: Transport,
)

Bind registry request specs to one configured transport.

Parameters

transportTransportrequired

Concrete transport owning URL, TLS, authentication, limits, retry policy, health, and metrics. No runtime type validation or cloning occurs in the constructor.

Attributes

transportTransport

The exact supplied object.

urlstr

Read-only proxy to transport.url.

RegistryClient.register

async methodprotolink.client.RegistryClient.register

source↗

async register(
  card: AgentCard,
) -> dict[str, str]

Serialize the card and POST it to /agents/ at the transport's own URL.

Parameters

cardAgentCardrequired

Card converted with to_dict() before transport dispatch.

Returns

statusdict[str, str]

Decoded handler payload.

Raises

transport or remote error

Connection, timeout, authentication, serialization, response, and server failures propagate.

RegistryClient.unregister

async methodprotolink.client.RegistryClient.unregister

source↗

async unregister(
  agent_url: str,
) -> dict[str, str]

Send an idempotent DELETE request with agent_url in its body.

Parameters

agent_urlstrrequired

Stable registry key to remove.

Returns

statusdict[str, str]

Decoded status payload.

RegistryClient.heartbeat

async methodprotolink.client.RegistryClient.heartbeat

source↗

async heartbeat(
  agent_url: str,
) -> dict[str, str]

Send an idempotent POST on the control-like liveness endpoint. The Registry request spec uses the default channel, so multiplexed transports do not automatically isolate heartbeats into "control".

Parameters

agent_urlstrrequired

Stable URL whose entry should remain live.

Returns

statusdict[str, str]

Recorded or not-found status.

RegistryClient.discover

async methodprotolink.client.RegistryClient.discover

source↗

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

Send optional filters as GET query parameters, then reconstruct every returned mapping as an AgentCard.

Parameters

filter_bydict[str, Any] | Nonedefault: None

Filter mapping or no query data.

Returns

cardslist[AgentCard]

Validated cards in server iteration order.

Raises

transport, shape, or AgentCard error

Request failures, a non-iterable response, or malformed card mappings propagate.

URL Handling

The transport is the URL source of truth. RegistryClient.url exposes it directly; Registry exposes the client through registry.client, so use registry.client.url when code needs the resolved public address. The current Registry class does not define a direct registry.url property.

from protolink.transport import HTTPTransport

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

assert registry.client.url == transport.url

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