NEW registry transport, server and status

Author: nMaroulis

README.md

@@ -88,12 +88,6 @@ Protolink takes a **centralized agent** approach compared to Google's A2A protoc
    - Simple interface-based design
    - No complex configuration needed for common use cases
 
-Once the Agent has been initiated, it automatically exposes a web interface at `/status` where it exposes the agent's information.
-
-<div align="center">
-  <img src="https://raw.githubusercontent.com/nMaroulis/protolink/main/docs/assets/agent_status_card.png" alt="Agent Status Card" width="50%">
-</div>
-
 ## Why Protolink? 🚀
 - **Real Multi-Agent Systems**: Build **autonomous agents** with embedded LLMs, tools, and memory that communicate directly.
 - **Simple API**: Built from the ground-up for **minimal boilerplate**, letting you focus on agent logic rather than infrastructure.
@@ -142,7 +136,7 @@ cd protolink
 uv pip install -e ".[dev]"
 ```
 
-## Quick Start
+## Hello World Example
 
 ```python
 from protolink.agents import Agent
@@ -183,6 +177,12 @@ agent.add_tool(mcp_tool)
 agent.start()
 ```
 
+Once the Agent has been initiated, it automatically exposes a web interface at `/status` where it exposes the agent's information.
+
+<div align="center">
+  <img src="https://raw.githubusercontent.com/nMaroulis/protolink/main/docs/assets/agent_status_card.png" alt="Agent Status Card" width="50%">
+</div>
+
 ## Documentation
 
 Follow the API documentation here: [Documentation](https://nmaroulis.github.io/protolink/)

examples/notebooks/agent_registry_test/registry.ipynb

@@ -31,8 +31,7 @@
     "registry = Registry(transport=transport)\n",
     "\n",
     "# Start registry\n",
-    "await registry.start()\n",
-    "print(\"Registry is running...\")"
+    "await registry.start()"
    ]
   }
  ],

examples/notebooks/agent_registry_test/weather_agent.ipynb

@@ -40,22 +40,15 @@
     "        return task.complete(f\"Weather data: {result}\")\n",
     "\n",
     "\n",
-    "# -------------------------\n",
-    "# Transports\n",
-    "# -------------------------\n",
-    "\n",
     "# A2A Transport\n",
     "transport = HTTPAgentTransport(url=URL)\n",
     "\n",
     "# Define Agent Card using the Agent Card class\n",
     "card = AgentCard(url=URL, name=\"WeatherAgent\", description=\"Produces weather data\")\n",
     "\n",
+    "# Initialize the agent\n",
     "agent = WeatherAgent(card=card, transport=transport, registry=REGISTRY_URL)\n",
     "\n",
-    "# -------------------------\n",
-    "# Tools\n",
-    "# -------------------------\n",
-    "\n",
     "\n",
     "# Add Native tool using the decorator\n",
     "@agent.tool(name=\"get_weather\", description=\"Return weather data for a city\")\n",
@@ -64,7 +57,7 @@
     "    return {\"city\": city, \"temperature\": 28, \"condition\": \"sunny\"}\n",
     "\n",
     "\n",
-    "await agent.start(register=False)"
+    "await agent.start(register=True)"
    ]
   },
   {

protolink/agents/base.py

@@ -102,7 +102,7 @@ def __init__(
         self.start_time: float | None = None
 
     # ----------------------------------------------------------------------
-    # Agent Lifecycle Init - A2A Server Operations
+    # Agent Server Lifecycle - A2A Operations
     # ----------------------------------------------------------------------
 
     async def start(self, *, register: bool = True) -> None:

protolink/discovery/registry.py

@@ -1,10 +1,13 @@
 # protolink/registry/registry.py
+import time
 from typing import Any
 
 from protolink.client import RegistryClient
 from protolink.models import AgentCard
+from protolink.server import RegistryServer
 from protolink.transport import HTTPRegistryTransport, RegistryTransport
 from protolink.utils.logging import get_logger
+from protolink.utils.renderers import to_registry_status_html
 
 
 class Registry:
@@ -28,49 +31,66 @@ def __init__(self, transport: RegistryTransport | None = None, url: str | None =
 
         # Create default HTTP transport if none provided
         if transport is None:
-            transport = HTTPRegistryTransport(url=url)
-
-        self.transport = transport
-        self.client = RegistryClient(self.transport)
+            if url is None:
+                raise ValueError("At least one of transport or url must be provided")
+            else:
+                self.logger.info(f"Creating default HTTPRegistryTransport using the provided URL: {url}")
+                transport = HTTPRegistryTransport(url=url)
 
         # Local store for agent cards
         self._agents: dict[str, AgentCard] = {}
 
-        # Wire server-side handlers
-        self.transport._register_local = self._register_local
-        self.transport._unregister_local = self._unregister_local
-        self.transport._discover_local = self._discover_local
+        self.start_time: float | None = None
+
+        # Setup registry client
+        self._client = RegistryClient(transport)
+
+        # Setup registry server
+        self._server = RegistryServer(
+            transport,
+            register_handler=self.handle_register,
+            unregister_handler=self.handle_unregister,
+            discover_handler=self.handle_discover,
+            status_handler=self.handle_status_html,
+        )
 
     # ------------------------------------------------------------------
-    # Lifecycle
+    # Registry Server Lifecycle
     # ------------------------------------------------------------------
 
     async def start(self) -> None:
         """Start the registry server via the transport."""
-        await self.transport.start()
+        if self._server:
+            try:
+                await self._server.start()
+            except Exception as e:
+                self.logger.exception(f"Unexpected error during server start: {e}")
+                raise
+        self.start_time = time.time()
 
     async def stop(self) -> None:
         """Stop the registry server via the transport."""
-        await self.transport.stop()
+        if self._server:
+            await self._server.stop()
 
     # ------------------------------------------------------------------
     # Client API (agents call these)
     # ------------------------------------------------------------------
 
     async def register(self, card: AgentCard) -> None:
-        await self.client.register(card)
+        await self._client.register(card)
 
     async def unregister(self, agent_url: str) -> None:
-        await self.client.unregister(agent_url)
+        await self._client.unregister(agent_url)
 
     async def discover(self, filter_by: dict[str, Any] | None = None) -> list[AgentCard]:
-        return await self.client.discover(filter_by)
+        return await self._client.discover(filter_by)
 
     # ------------------------------------------------------------------
     # Server-side handlers
     # ------------------------------------------------------------------
 
-    async def _register_local(self, card: AgentCard) -> None:
+    async def handle_register(self, card: AgentCard) -> None:
         self._agents[card.url] = card
 
         self.logger.info(
@@ -81,10 +101,10 @@ async def _register_local(self, card: AgentCard) -> None:
             },
         )
 
-    async def _unregister_local(self, agent_url: str) -> None:
+    async def handle_unregister(self, agent_url: str) -> None:
         self._agents.pop(agent_url, None)
 
-    async def _discover_local(self, filter_by: dict[str, Any] | None = None) -> list[AgentCard]:
+    async def handle_discover(self, filter_by: dict[str, Any] | None = None) -> list[AgentCard]:
         if not filter_by:
             return list(self._agents.values())
 
@@ -93,6 +113,14 @@ def match(card: AgentCard) -> bool:
 
         return [c for c in self._agents.values() if match(c)]
 
+    def handle_status_html(self) -> str:
+        """Return the registry's status as HTML.
+
+        Returns:
+            HTML string with registry status information
+        """
+        return to_registry_status_html("Registry", "HTTP", self._agents, self.start_time)
+
     # ------------------------------------------------------------------
     # Utilities
     # ------------------------------------------------------------------

protolink/llms/api/openai_client.py

@@ -67,8 +67,6 @@ def _to_message(self, response: Any) -> Message:
         """Convert OpenAI completion to internal Message format."""
 
         output_text: str = ""
-        print(type(response))
-        print(response)
         for item in response.output or []:
             # item: ResponseOutputMessage
             if item.type != "message":

protolink/server/__init__.py

@@ -1,5 +1,7 @@
-from .agent_server import AgentServer
+from .agent import AgentServer
+from .registry import RegistryServer
 
 __all__ = [
     "AgentServer",
+    "RegistryServer",
 ]

protolink/server/agent_server.py protolink/server/agent.pyprotolink/server/agent_server.py renamed to protolink/server/agent.py

@@ -0,0 +1,90 @@
+"""Registry server implementation for handling incoming requests."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from protolink.models import AgentCard
+from protolink.transport import RegistryTransport
+
+
+class RegistryServer:
+    """Thin wrapper that wires a task handler into a transport."""
+
+    def __init__(
+        self,
+        transport: RegistryTransport,
+        register_handler: Callable[[AgentCard], Awaitable[None]] | None = None,
+        unregister_handler: Callable[[str], Awaitable[None]] | None = None,
+        discover_handler: Callable[[dict[str, Any]], Awaitable[list[AgentCard]]] | None = None,
+        status_handler: Callable[[], Awaitable[str]] | None = None,
+    ) -> None:
+        if transport is None:
+            raise ValueError("RegistryServer requires a transport instance")
+
+        self._transport = transport
+        self._register_handler = None
+        self._unregister_handler = None
+        self._discover_handler = None
+        self._status_handler = None
+        self._is_running = False
+
+        if register_handler is not None:
+            self.set_register_handler(register_handler)
+
+        if unregister_handler is not None:
+            self.set_unregister_handler(unregister_handler)
+
+        if discover_handler is not None:
+            self.set_discover_handler(discover_handler)
+
+        if status_handler is not None:
+            self.set_status_handler(status_handler)
+
+    def set_register_handler(self, handler: Callable[[AgentCard], Awaitable[None]]) -> None:
+        """Register the coroutine used to process incoming register requests from agents."""
+
+        self._register_handler = handler
+        self._transport.on_register_received(handler)
+
+    def set_unregister_handler(self, handler: Callable[[str], Awaitable[None]]) -> None:
+        """Register the coroutine used to process incoming unregister requests from agents."""
+
+        self._unregister_handler = handler
+        self._transport.on_unregister_received(handler)
+
+    def set_discover_handler(self, handler: Callable[[dict[str, Any]], Awaitable[list[AgentCard]]]) -> None:
+        """Register the coroutine used to process incoming discover requests from agents."""
+
+        self._discover_handler = handler
+        self._transport.on_discover_received(handler)
+
+    def set_status_handler(self, handler: Callable[[], Awaitable[str]]) -> None:
+        """Register the coroutine used to process incoming status requests from agents."""
+        self._status_handler = handler
+        self._transport.on_status_received(handler)
+
+    async def start(self) -> None:
+        """Start the underlying transport."""
+
+        if self._is_running:
+            return
+
+        if not self._register_handler:
+            raise RuntimeError("No register handler registered. Call set_register_handler() first.")
+
+        if not self._unregister_handler:
+            raise RuntimeError("No unregister handler registered. Call set_unregister_handler() first.")
+
+        await self._transport.start()
+        self._is_running = True
+
+    async def stop(self) -> None:
+        """Stop the underlying transport and mark the server as idle."""
+
+        if not self._is_running:
+            return
+
+        await self._transport.stop()
+        self._is_running = False

protolink/server/registry.py

@@ -36,8 +36,7 @@ def setup_routes(self, transport: "HTTPAgentTransport") -> None:  # noqa: F821
         """Register all HTTP routes on the Starlette application.
 
         This method wires the public HTTP API to the internal transport handlers.
-        Each route is registered via a dedicated helper for clarity and separation
-        of concerns.
+        Each route is registered via a dedicated helper for clarity and separation of concerns.
         """
         self._setup_task_routes(transport)
         self._setup_agent_card_routes(transport)