zelos.protocol_adapters

  1"""
  2Phase 2 Protocol Adapters — gRPC, MCP, A2A, WebSocket.
  3
  4Each adapter translates an external protocol → Runtime API.
  5Adapters contain ZERO business logic. They are stateless, replaceable plugins.
  6
  7Architecture:
  8  Protocol Adapter (translate) → Runtime API → Kernel
  9"""
 10
 11import threading
 12import time
 13from abc import ABC, abstractmethod
 14from typing import Any
 15
 16# ═══════════════════ Adapter Base ═══════════════════
 17
 18
 19class ProtocolAdapter(ABC):
 20    """Base for all protocol adapters."""
 21
 22    def __init__(self, runtime=None):
 23        self.runtime = runtime
 24
 25    @abstractmethod
 26    def start(self) -> None: ...
 27
 28    @abstractmethod
 29    def stop(self) -> None: ...
 30
 31
 32# ═══════════════════ gRPC Adapter ═══════════════════
 33
 34
 35class GRPCAdapter(ProtocolAdapter):
 36    """
 37    gRPC Adapter — translates gRPC calls to Runtime API.
 38
 39    Proto service definition (zelos.proto):
 40
 41      service Zelos {
 42        rpc SubmitGoal(SubmitGoalRequest) returns (GoalResponse);
 43        rpc GetGoalStatus(GetGoalStatusRequest) returns (GoalStatusResponse);
 44        rpc CancelGoal(CancelGoalRequest) returns (GoalStatusResponse);
 45        rpc WatchGoal(WatchGoalRequest) returns (stream Event);
 46        rpc RegisterAgent(RegisterAgentRequest) returns (AgentResponse);
 47        rpc AgentHeartbeat(HeartbeatRequest) returns (HeartbeatResponse);
 48        rpc SubmitResult(SubmitResultRequest) returns (SubmitResultResponse);
 49        rpc GetHealth(Empty) returns (HealthResponse);
 50        rpc GetMetrics(Empty) returns (MetricsResponse);
 51      }
 52
 53    Phase 2: Full service handler implementation. For production,
 54    use grpcio to create the server with generated stubs.
 55    """
 56
 57    def __init__(self, runtime=None, host: str = "0.0.0.0", port: int = 50051):
 58        super().__init__(runtime)
 59        self.host = host
 60        self.port = port
 61        self._running = False
 62
 63    def start(self) -> None:
 64        self._running = True
 65        # In production: grpc.server with generated ZelosServicer
 66        # Phase 2 reference: service handlers below
 67
 68    def stop(self) -> None:
 69        self._running = False
 70
 71    # ── Service Handlers ──
 72
 73    def SubmitGoal(self, request: dict) -> dict:
 74        """gRPC → Runtime API: SubmitGoal."""
 75        if not self.runtime:
 76            return {"goal_id": "", "status": "rejected", "reason": "Runtime not connected"}
 77        return self.runtime.submit_goal(
 78            description=request.get("description", ""),
 79            budget=request.get("budget"),
 80            deadline=request.get("deadline"),
 81            priority=request.get("priority", "medium"),
 82        )
 83
 84    def GetGoalStatus(self, request: dict) -> dict:
 85        if not self.runtime:
 86            return {"goal_id": "", "status": "not_found"}
 87        return self.runtime.get_goal_status(request["goal_id"]) or {"status": "not_found"}
 88
 89    def CancelGoal(self, request: dict) -> dict:
 90        if not self.runtime:
 91            return {}
 92        return self.runtime.cancel_goal(request["goal_id"]) or {}
 93
 94    def RegisterAgent(self, request: dict) -> dict:
 95        if not self.runtime:
 96            return {"agent_id": "", "status": "rejected"}
 97        agent_id = self.runtime.add_agent(
 98            name=request.get("name", "grpc-agent"),
 99            entrypoint=request.get("entrypoint", ""),
100            capabilities=request.get("capabilities", []),
101        )
102        return {
103            "agent_id": agent_id,
104            "status": "registered",
105            "heartbeat_interval_ms": 30000,
106            "runtime_version": "0.7.0",
107        }
108
109    def AgentHeartbeat(self, request: dict) -> dict:
110        if not self.runtime:
111            return {"status": "re-register"}
112        ok = self.runtime._execution_engine.heartbeat(request["agent_id"])
113        return {"status": "ok" if ok else "re-register", "pending_tasks": 0}
114
115    def SubmitResult(self, request: dict) -> dict:
116        if not self.runtime:
117            return {"status": "rejected"}
118        ok = self.runtime._execution_engine.submit_result(
119            request.get("task_id", ""), request.get("agent_id", ""), request.get("result", {})
120        )
121        return {"status": "accepted" if ok else "rejected"}
122
123    def GetHealth(self, _=None) -> dict:
124        return self.runtime.get_health() if self.runtime else {"status": "unhealthy"}
125
126    def GetMetrics(self, _=None) -> dict:
127        return self.runtime.get_metrics() if self.runtime else {}
128
129
130# ═══════════════════ WebSocket Adapter ═══════════════════
131
132
133class WebSocketAdapter(ProtocolAdapter):
134    """
135    WebSocket Adapter — streams Events to connected clients.
136
137    Supports:
138      - watch_goal(goal_id): stream all events for a Goal
139      - watch_tasks(pattern): stream task.* events
140      - health streaming
141
142    Phase 2: Reference implementation using a simple event subscription model.
143    """
144
145    def __init__(self, runtime=None):
146        super().__init__(runtime)
147        self._clients: dict[str, dict] = {}  # client_id → {subscriptions, queue}
148        self._running = False
149        self._thread: threading.Thread | None = None
150
151    def start(self) -> None:
152        self._running = True
153        if self.runtime:
154            # Subscribe to all major event domains
155            for domain in ["goal", "task", "plan", "agent", "artifact", "verification", "plugin"]:
156                self.runtime._event_bus.subscribe_pattern(f"{domain}.*", self._on_event)
157
158    def stop(self) -> None:
159        self._running = False
160        self._clients.clear()
161
162    def register_client(self, client_id: str) -> None:
163        self._clients[client_id] = {"subscriptions": [], "queue": []}
164
165    def unregister_client(self, client_id: str) -> None:
166        self._clients.pop(client_id, None)
167
168    def watch_goal(self, client_id: str, goal_id: str) -> None:
169        if client_id in self._clients:
170            self._clients[client_id]["subscriptions"].append(f"goal.{goal_id}")
171
172    def get_events(self, client_id: str) -> list[dict]:
173        """Drain event queue for a client."""
174        if client_id not in self._clients:
175            return []
176        queue = self._clients[client_id]["queue"]
177        events = list(queue)
178        queue.clear()
179        return events
180
181    def _on_event(self, event) -> Any:
182        """Fan out event to interested clients."""
183        for _cid, client in self._clients.items():
184            for sub in client["subscriptions"]:
185                # Simple match: "goal.g-001" matches events with correlation_id
186                parts = sub.split(".", 1)
187                if len(parts) == 2 and event.correlation_id == parts[1]:
188                    client["queue"].append(event.to_dict())
189                elif event.event_type.startswith(sub):
190                    client["queue"].append(event.to_dict())
191        from zelos.event_bus import HandlerResult
192
193        return HandlerResult.ACK
194
195
196# ═══════════════════ MCP Adapter ═══════════════════
197
198
199class MCPAdapter(ProtocolAdapter):
200    """
201    MCP (Model Context Protocol) Adapter.
202
203    MCP is how Agents access TOOLS during Task execution — NOT how they
204    communicate with the Runtime. The Runtime is unaware of MCP internals.
205
206    Phase 2: MCP Server lifecycle management + Tool registry.
207    """
208
209    def __init__(self, runtime=None):
210        super().__init__(runtime)
211        self._tool_registry: dict[str, dict] = {}  # tool_name → {server, schema}
212
213    def start(self) -> None: ...
214
215    def stop(self) -> None: ...
216
217    def register_tool(self, name: str, server_endpoint: str, schema: dict) -> None:
218        """Register an MCP tool available to Agents."""
219        self._tool_registry[name] = {"endpoint": server_endpoint, "schema": schema}
220
221    def list_tools(self) -> list[dict]:
222        """Return all registered MCP tools (MCP tools/list format)."""
223        return [
224            {
225                "name": name,
226                "description": info["schema"].get("description", ""),
227                "inputSchema": info["schema"].get("input_schema", {}),
228            }
229            for name, info in self._tool_registry.items()
230        ]
231
232    def call_tool(self, name: str, arguments: dict) -> dict:
233        """Invoke an MCP tool (MCP tools/call format)."""
234        tool = self._tool_registry.get(name)
235        if not tool:
236            return {"isError": True, "content": [{"type": "text", "text": f"Tool not found: {name}"}]}
237        # Phase 2: In production, forward to the tool server
238        return {"content": [{"type": "text", "text": f"Tool '{name}' invoked with {arguments}"}]}
239
240
241# ═══════════════════ A2A Adapter ═══════════════════
242
243
244class A2AAdapter(ProtocolAdapter):
245    """
246    A2A (Agent-to-Agent Protocol) Adapter.
247
248    A2A is an EXTERNAL interoperability protocol for:
249      - Runtime ↔ Runtime coordination
250      - External Agent integration
251      - Third-party Agent services
252
253    Key translation: A2A Agent Card ↔ Zelos Capability list
254                     A2A Task ↔ Zelos Task
255                     A2A Message ↔ Zelos Event
256
257    A2A assumes direct agent-to-agent communication. Zelos PROHIBITS this.
258    The A2A Adapter routes ALL A2A messages through the Runtime:
259      A2A: Agent A → Agent B
260      → Zelos: Agent A → Runtime → Agent B
261    """
262
263    def __init__(self, runtime=None):
264        super().__init__(runtime)
265        self._external_agents: dict[str, dict] = {}
266
267    def start(self) -> None: ...
268
269    def stop(self) -> None: ...
270
271    def generate_agent_card(self, agent_id: str) -> dict:
272        """Generate A2A Agent Card from Zelos Agent Capabilities."""
273        if not self.runtime:
274            return {}
275        agent = self.runtime.get_agent(agent_id)
276        if not agent:
277            return {}
278        return {
279            "agentId": agent.get("agent_id", ""),
280            "name": agent.get("name", ""),
281            "description": f"Zelos Agent providing {len(agent.get('capabilities', []))} capabilities",
282            "skills": [
283                {"name": c["name"], "description": c.get("description", "")} for c in agent.get("capabilities", [])
284            ],
285            "endpoint": f"zelos://agents/{agent.get('agent_id', '')}",
286        }
287
288    def receive_external_task(self, task_data: dict) -> str | None:
289        """Receive an A2A Task from an external system → create Zelos Task."""
290        if not self.runtime:
291            return None
292        goal = self.runtime.submit_goal(
293            description=task_data.get("description", "External A2A task"),
294            priority=task_data.get("priority", "medium"),
295        )
296        return goal.get("goal_id")
297
298    def register_external_agent(self, agent_card: dict) -> str:
299        """Register an external (non-Zelos) Agent via its A2A Card."""
300        import uuid
301
302        aid = str(uuid.uuid4())
303        self._external_agents[aid] = {
304            "card": agent_card,
305            "registered_at": time.time(),
306        }
307        return aid