diff --git a/clawloop/environments/_car_purple.py b/clawloop/environments/_car_purple.py index 22a8c073..7060f46f 100644 --- a/clawloop/environments/_car_purple.py +++ b/clawloop/environments/_car_purple.py @@ -3,27 +3,36 @@ from __future__ import annotations -import asyncio -import json import logging -import threading -from typing import Any +from typing import Any, ClassVar from uuid import uuid4 -import litellm -import uvicorn -from starlette.applications import Starlette -from starlette.requests import Request -from starlette.responses import JSONResponse -from starlette.routing import Route - +from clawloop.environments._purple_base import ( + _PurpleAgentBase, + create_app, + start_purple_server, +) from clawloop.learning_layers.harness import Harness log = logging.getLogger(__name__) +__all__ = ["CarPurpleAgent", "create_app", "start_purple_server"] + + +class CarPurpleAgent(_PurpleAgentBase): + """Purple agent for CAR-bench: parses System/User prompt, wraps replies in ``message``.""" -class CarPurpleAgent: - """A2A-compliant purple agent that injects clawloop harness state into LLM calls.""" + default_bench: ClassVar[str] = "car" + agent_card_name: ClassVar[str] = "clawloop-purple-agent" + agent_card_description: ClassVar[str] = "ClawLoop harness-optimized agent under test" + agent_card_skills: ClassVar[list[dict]] = [ + { + "id": "car_assistant", + "name": "In-Car Voice Assistant", + "description": "Agent under test for CAR-bench evaluation", + "tags": ["benchmark", "car-bench"], + } + ] def __init__( self, @@ -33,28 +42,16 @@ def __init__( api_base: str | None = None, api_key: str | None = None, ): - self.model = model - self.harness = harness - self.bench = bench - self.api_base = api_base - self.api_key = api_key - self._sessions: dict[str, list[dict]] = {} - self._tool_cache: dict[str, list[dict]] = {} + super().__init__(model, harness, bench, api_base, api_key) self._captured: dict[str, list[dict]] = {} - def update_harness(self, harness: Harness) -> None: - self.harness = harness - def clear_all_sessions(self) -> None: - self._sessions.clear() - self._tool_cache.clear() + super().clear_all_sessions() self._captured.clear() - # -- Message parsing -- - @staticmethod def _parse_first_message(raw_text: str) -> tuple[str, str]: - """Parse 'System: ...\\n\\nUser: ...' format from green agent.""" + """Parse ``System: ...\\n\\nUser: ...`` format from green agent.""" if "System:" in raw_text and "\n\nUser:" in raw_text: parts = raw_text.split("\n\nUser:", 1) system = parts[0].replace("System:", "", 1).strip() @@ -62,281 +59,27 @@ def _parse_first_message(raw_text: str) -> tuple[str, str]: return system, user return "", raw_text - @staticmethod - def _convert_tools_to_openai(car_tools: list[dict]) -> list[dict]: - """Normalize tool schemas to OpenAI function-calling format. + def _build_initial_messages(self, text_parts: list[str]) -> list[dict]: + raw_text = text_parts[0] if text_parts else "" + system_prompt, user_text = self._parse_first_message(raw_text) - Green agent may send tools already wrapped ({type: function, function: ...}) - or flat ({name, description, parameters}). Handle both. - """ - result = [] - for t in car_tools: - if t.get("type") == "function" and "function" in t: - # Already in OpenAI format - result.append(t) - else: - result.append( - { - "type": "function", - "function": { - "name": t["name"], - "description": t.get("description", ""), - "parameters": t.get("parameters", {}), - }, - } - ) - return result + harness_prompt = self.harness.system_prompt(self.bench) + if harness_prompt: + system_prompt = f"{harness_prompt}\n\n{system_prompt}" - @staticmethod - def _normalize_assistant_msg(litellm_msg: Any) -> dict: - """Normalize litellm response to stable internal format.""" - normalized: dict[str, Any] = { - "role": "assistant", - "content": litellm_msg.content or "", - } - if litellm_msg.tool_calls: - normalized["tool_calls"] = [ - { - "id": tc.id, - "type": "function", - "function": { - "name": tc.function.name, - "arguments": tc.function.arguments, - }, - } - for tc in litellm_msg.tool_calls - ] - return normalized + out: list[dict] = [{"role": "system", "content": system_prompt}] + if user_text: + out.append({"role": "user", "content": user_text}) + return out def _format_a2a_response(self, assistant_msg: Any) -> dict: - """Format LLM response as A2A result body.""" - parts: list[dict] = [{"kind": "text", "text": assistant_msg.content or ""}] - - if assistant_msg.tool_calls: - tool_calls = [] - for tc in assistant_msg.tool_calls: - args = tc.function.arguments - if args is None: - args = {} - elif isinstance(args, str): - try: - args = json.loads(args) - except json.JSONDecodeError: - log.warning("Malformed tool args for %s", tc.function.name) - args = {"raw": args} - tool_calls.append({"tool_name": tc.function.name, "arguments": args}) - parts.append({"kind": "data", "data": {"tool_calls": tool_calls}}) - return { "message": { "messageId": uuid4().hex, "role": "agent", - "parts": parts, + "parts": self._build_message_parts(assistant_msg), } } - # -- Core message handling -- - - def handle_message_sync(self, jsonrpc_request: dict) -> dict: - """Handle one message/send request (sync — litellm.completion is sync).""" - params = jsonrpc_request["params"] - msg = params["message"] - context_id = params.get("contextId", "default") - - text_parts = [p["text"] for p in msg["parts"] if p.get("kind") == "text"] - data_parts = [p["data"] for p in msg["parts"] if p.get("kind") == "data"] - - # Initialize session - if context_id not in self._sessions: - self._sessions[context_id] = [] - self._captured[context_id] = [] - - messages = self._sessions[context_id] - - if not messages: - # First message: extract system prompt + tools - raw_text = text_parts[0] if text_parts else "" - system_prompt, user_text = self._parse_first_message(raw_text) - - # HARNESS INJECTION - harness_prompt = self.harness.system_prompt(self.bench) - if harness_prompt: - system_prompt = f"{harness_prompt}\n\n{system_prompt}" - - messages.append({"role": "system", "content": system_prompt}) - if user_text: - messages.append({"role": "user", "content": user_text}) - - # Cache tools - for d in data_parts: - if "tools" in d: - self._tool_cache[context_id] = self._convert_tools_to_openai(d["tools"]) - else: - # Subsequent: tool results and/or user text - for d in data_parts: - if "tool_results" in d: - for tr in d["tool_results"]: - # Reconcile tool_call_ids: rewrite last assistant msg's - # tool_call ids to match green's generated ids - green_id = tr["tool_call_id"] - tool_name = tr.get("tool_name", "") - self._reconcile_tool_call_id(messages, tool_name, green_id) - - messages.append( - { - "role": "tool", - "tool_call_id": green_id, - "content": tr["content"], - } - ) - for text in text_parts: - if text.strip(): - messages.append({"role": "user", "content": text}) - - # Call LLM - tools = self._tool_cache.get(context_id) - completion_kwargs: dict[str, Any] = { - "model": self.model, - "messages": messages, - "temperature": 0.0, - } - if tools: - completion_kwargs["tools"] = tools - if self.api_base: - completion_kwargs["api_base"] = self.api_base - if self.api_key: - completion_kwargs["api_key"] = self.api_key - - response = litellm.completion(**completion_kwargs) - assistant_msg = response.choices[0].message - - # Normalize and store - normalized = self._normalize_assistant_msg(assistant_msg) - messages.append(normalized) - self._captured[context_id].append(normalized) - - return self._format_a2a_response(assistant_msg) - - @staticmethod - def _reconcile_tool_call_id(messages: list[dict], tool_name: str, green_id: str) -> None: - """Rewrite last assistant message's tool_call id to match green's id. - - Green generates its own tool_call_ids. The LLM needs matching ids between - assistant tool_calls and tool-role messages. We rewrite the assistant msg's - id to match what green sent back. - - Handles duplicate tool names by only rewriting tool calls that haven't - already been reconciled (i.e., still have their original LLM-generated id). - """ - # Collect green IDs already used in existing tool-role messages - used_green_ids = { - m["tool_call_id"] for m in messages if m.get("role") == "tool" and "tool_call_id" in m - } - # Walk backwards to find the last assistant message with tool_calls - for msg in reversed(messages): - if msg.get("role") != "assistant" or "tool_calls" not in msg: - continue - for tc in msg["tool_calls"]: - # Match by name, skip if already reconciled (id is a known green id) - if tc["function"]["name"] == tool_name and tc["id"] not in used_green_ids: - tc["id"] = green_id - return - return # found assistant msg but no matching tool name - - -def create_app(agent: CarPurpleAgent, port: int = 0) -> Starlette: - """Create the A2A Starlette app.""" - - async def agent_card(request: Request) -> JSONResponse: - return JSONResponse( - { - "name": "clawloop-purple-agent", - "description": "ClawLoop harness-optimized agent under test", - "url": f"http://127.0.0.1:{port}/", - "version": "0.1.0", - "protocol_version": "0.3.0", - "preferred_transport": "JSONRPC", - "default_input_modes": ["text/plain"], - "default_output_modes": ["text/plain"], - "capabilities": {"streaming": False, "push_notifications": False}, - "skills": [ - { - "id": "car_assistant", - "name": "In-Car Voice Assistant", - "description": "Agent under test for CAR-bench evaluation", - "tags": ["benchmark", "car-bench"], - } - ], - } - ) - - async def handle_jsonrpc(request: Request) -> JSONResponse: - body = await request.json() - if body.get("jsonrpc") != "2.0" or "id" not in body: - return JSONResponse( - { - "jsonrpc": "2.0", - "id": None, - "error": {"code": -32600, "message": "Invalid Request"}, - } - ) - - method = body.get("method") - if method != "message/send": - return JSONResponse( - { - "jsonrpc": "2.0", - "id": body["id"], - "error": {"code": -32601, "message": f"Method not found: {method}"}, - } - ) - - # Run sync litellm call in thread to avoid blocking event loop - loop = asyncio.get_running_loop() - result = await loop.run_in_executor(None, agent.handle_message_sync, body) - - return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result}) - - return Starlette( - routes=[ - Route("/.well-known/agent.json", agent_card, methods=["GET"]), - Route("/.well-known/agent-card.json", agent_card, methods=["GET"]), - Route("/", handle_jsonrpc, methods=["POST"]), - ] - ) - - -def start_purple_server( - agent: CarPurpleAgent, host: str = "127.0.0.1", port: int = 0 -) -> tuple[threading.Thread, int]: - """Start the purple agent server in a background thread. Returns (thread, actual_port).""" - import socket - import time - - # Bind socket to get a free port; pass it directly to uvicorn to avoid race. - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.bind((host, port)) - actual_port = sock.getsockname()[1] - - app = create_app(agent, actual_port) - config = uvicorn.Config(app, host=host, port=actual_port, log_level="warning") - server = uvicorn.Server(config) - - thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True) - thread.start() - - # Poll for readiness - import httpx - - for _ in range(50): - try: - r = httpx.get(f"http://{host}:{actual_port}/.well-known/agent-card.json", timeout=0.5) - if r.status_code == 200: - break - except httpx.ConnectError: - time.sleep(0.1) - else: - log.warning("Purple server did not become ready within 5s") - - return thread, actual_port + def _capture_assistant(self, context_id: str, normalized: dict) -> None: + self._captured.setdefault(context_id, []).append(normalized) diff --git a/clawloop/environments/_entropic_purple.py b/clawloop/environments/_entropic_purple.py index e7e4977f..5b5ed824 100644 --- a/clawloop/environments/_entropic_purple.py +++ b/clawloop/environments/_entropic_purple.py @@ -1,134 +1,38 @@ # clawloop/environments/_entropic_purple.py -"""A2A purple agent server for Entropic CRMArenaPro with clawloop harness injection. - -The entropic green agent sends CRM task prompts as plain text via A2A -``message/send``. This purple agent: - 1. Injects the clawloop harness system prompt. - 2. Forwards the task to the configured LLM. - 3. Returns the answer as an A2A text response. - -Multi-turn is supported: the green agent may send follow-up messages when the -agent requests clarification or when tool results are returned. -""" +"""A2A purple agent server for Entropic CRMArenaPro with clawloop harness injection.""" from __future__ import annotations -import asyncio import json import logging -import threading -from typing import Any +from typing import Any, ClassVar from uuid import uuid4 -import litellm -import uvicorn -from starlette.applications import Starlette -from starlette.requests import Request -from starlette.responses import JSONResponse -from starlette.routing import Route - -from clawloop.learning_layers.harness import Harness +from clawloop.environments._purple_base import ( + _PurpleAgentBase, + create_app, + start_purple_server, +) log = logging.getLogger(__name__) +__all__ = ["EntropicPurpleAgent", "create_app", "start_purple_server"] -class EntropicPurpleAgent: - """A2A-compliant purple agent for CRM tasks with clawloop harness injection.""" - - def __init__( - self, - model: str, - harness: Harness, - bench: str = "entropic", - api_base: str | None = None, - api_key: str | None = None, - ): - self.model = model - self.harness = harness - self.bench = bench - self.api_base = api_base - self.api_key = api_key - self._sessions: dict[str, list[dict]] = {} - self._tool_cache: dict[str, list[dict]] = {} - def update_harness(self, harness: Harness) -> None: - self.harness = harness - - def clear_all_sessions(self) -> None: - self._sessions.clear() - self._tool_cache.clear() - - # -- Tool schema helpers -- - - @staticmethod - def _convert_tools_to_openai(raw_tools: list[dict]) -> list[dict]: - """Normalize tool schemas to OpenAI function-calling format.""" - result = [] - for t in raw_tools: - if t.get("type") == "function" and "function" in t: - result.append(t) - else: - result.append( - { - "type": "function", - "function": { - "name": t["name"], - "description": t.get("description", ""), - "parameters": t.get("parameters", {}), - }, - } - ) - return result +class EntropicPurpleAgent(_PurpleAgentBase): + """Purple agent for Entropic CRMArenaPro: formats CRM task, flat message envelope.""" - @staticmethod - def _normalize_assistant_msg(litellm_msg: Any) -> dict: - """Normalize litellm response to stable internal dict.""" - normalized: dict[str, Any] = { - "role": "assistant", - "content": litellm_msg.content or "", - } - if litellm_msg.tool_calls: - normalized["tool_calls"] = [ - { - "id": tc.id, - "type": "function", - "function": { - "name": tc.function.name, - "arguments": tc.function.arguments, - }, - } - for tc in litellm_msg.tool_calls - ] - return normalized - - def _format_a2a_response(self, assistant_msg: Any) -> dict: - """Format LLM response as A2A result body.""" - parts: list[dict] = [{"kind": "text", "text": assistant_msg.content or ""}] - - if assistant_msg.tool_calls: - tool_calls = [] - for tc in assistant_msg.tool_calls: - args = tc.function.arguments - if args is None: - args = {} - elif isinstance(args, str): - try: - args = json.loads(args) - except json.JSONDecodeError: - log.warning("Malformed tool args for %s", tc.function.name) - args = {"raw": args} - tool_calls.append({"tool_name": tc.function.name, "arguments": args}) - parts.append({"kind": "data", "data": {"tool_calls": tool_calls}}) - - # Return Message directly (not wrapped) — a2a-sdk expects result=Message - return { - "kind": "message", - "messageId": uuid4().hex, - "role": "agent", - "parts": parts, + default_bench: ClassVar[str] = "entropic" + agent_card_name: ClassVar[str] = "clawloop-entropic-purple-agent" + agent_card_description: ClassVar[str] = "ClawLoop harness-optimized CRM agent under test" + agent_card_skills: ClassVar[list[dict]] = [ + { + "id": "crm_assistant", + "name": "CRM Assistant", + "description": "Agent under test for Entropic CRMArenaPro evaluation", + "tags": ["benchmark", "entropic", "crmarena"], } - - # -- CRM task formatting -- + ] @staticmethod def _format_crm_task(raw_text: str) -> str: @@ -136,12 +40,10 @@ def _format_crm_task(raw_text: str) -> str: The green agent sends ``json.dumps(task_context)`` as an A2A TextPart. We extract the structured fields and present them clearly to the LLM. - If the text isn't valid JSON, return it unchanged. + If the text isn't valid JSON (or lacks ``prompt``), return it unchanged. """ - import json as _json - try: - ctx = _json.loads(raw_text) + ctx = json.loads(raw_text) except (ValueError, TypeError): return raw_text @@ -178,10 +80,8 @@ def _format_crm_task(raw_text: str) -> str: @staticmethod def _extract_task_tags(raw_text: str) -> set[str] | None: """Extract task category from CRM task JSON for selective playbook retrieval.""" - import json as _json - try: - ctx = _json.loads(raw_text) + ctx = json.loads(raw_text) except (ValueError, TypeError): return None if not isinstance(ctx, dict): @@ -192,188 +92,23 @@ def _extract_task_tags(raw_text: str) -> set[str] | None: tags.add(cat) return tags or None - # -- Core message handling -- - - def handle_message_sync(self, jsonrpc_request: dict) -> dict: - """Handle one message/send request (sync).""" - params = jsonrpc_request["params"] - msg = params["message"] - context_id = params.get("contextId", "default") - - text_parts = [p["text"] for p in msg["parts"] if p.get("kind") == "text"] - data_parts = [p["data"] for p in msg["parts"] if p.get("kind") == "data"] - - # Initialize session - if context_id not in self._sessions: - self._sessions[context_id] = [] - - messages = self._sessions[context_id] - - if not messages: - # First message — inject harness as system prompt. - # Extract task tags for selective playbook retrieval (ACE/DC-RS). - raw_text = "\n".join(text_parts) - task_tags = self._extract_task_tags(raw_text) - harness_prompt = self.harness.system_prompt(self.bench, task_tags=task_tags) - system_content = harness_prompt or "You are a helpful CRM assistant." - messages.append({"role": "system", "content": system_content}) - - user_text = raw_text - user_text = self._format_crm_task(user_text) - if user_text.strip(): - messages.append({"role": "user", "content": user_text}) - - # Cache tools if provided - for d in data_parts: - if "tools" in d: - self._tool_cache[context_id] = self._convert_tools_to_openai(d["tools"]) - else: - # Subsequent turns — tool results and/or user text - for d in data_parts: - if "tool_results" in d: - for tr in d["tool_results"]: - green_id = tr["tool_call_id"] - tool_name = tr.get("tool_name", "") - self._reconcile_tool_call_id(messages, tool_name, green_id) - messages.append( - { - "role": "tool", - "tool_call_id": green_id, - "content": tr["content"], - } - ) - for text in text_parts: - if text.strip(): - messages.append({"role": "user", "content": text}) - - # Call LLM - tools = self._tool_cache.get(context_id) - completion_kwargs: dict[str, Any] = { - "model": self.model, - "messages": messages, - "temperature": 0.0, - } - if tools: - completion_kwargs["tools"] = tools - if self.api_base: - completion_kwargs["api_base"] = self.api_base - if self.api_key: - completion_kwargs["api_key"] = self.api_key - - response = litellm.completion(**completion_kwargs) - assistant_msg = response.choices[0].message + def _build_initial_messages(self, text_parts: list[str]) -> list[dict]: + raw_text = "\n".join(text_parts) + task_tags = self._extract_task_tags(raw_text) + harness_prompt = self.harness.system_prompt(self.bench, task_tags=task_tags) + system_content = harness_prompt or "You are a helpful CRM assistant." - normalized = self._normalize_assistant_msg(assistant_msg) - messages.append(normalized) + out: list[dict] = [{"role": "system", "content": system_content}] + user_text = self._format_crm_task(raw_text) + if user_text.strip(): + out.append({"role": "user", "content": user_text}) + return out - return self._format_a2a_response(assistant_msg) - - @staticmethod - def _reconcile_tool_call_id(messages: list[dict], tool_name: str, green_id: str) -> None: - """Rewrite last assistant tool_call id to match green's id.""" - used_green_ids = { - m["tool_call_id"] for m in messages if m.get("role") == "tool" and "tool_call_id" in m + def _format_a2a_response(self, assistant_msg: Any) -> dict: + # Return Message directly (not wrapped) — a2a-sdk expects result=Message + return { + "kind": "message", + "messageId": uuid4().hex, + "role": "agent", + "parts": self._build_message_parts(assistant_msg), } - for msg in reversed(messages): - if msg.get("role") != "assistant" or "tool_calls" not in msg: - continue - for tc in msg["tool_calls"]: - if tc["function"]["name"] == tool_name and tc["id"] not in used_green_ids: - tc["id"] = green_id - return - return - - -def create_app(agent: EntropicPurpleAgent, port: int = 0) -> Starlette: - """Create the A2A Starlette app for the entropic purple agent.""" - - async def agent_card(request: Request) -> JSONResponse: - return JSONResponse( - { - "name": "clawloop-entropic-purple-agent", - "description": "ClawLoop harness-optimized CRM agent under test", - "url": f"http://127.0.0.1:{port}/", - "version": "0.1.0", - "protocol_version": "0.3.0", - "preferred_transport": "JSONRPC", - "default_input_modes": ["text/plain"], - "default_output_modes": ["text/plain"], - "capabilities": {"streaming": False, "push_notifications": False}, - "skills": [ - { - "id": "crm_assistant", - "name": "CRM Assistant", - "description": "Agent under test for Entropic CRMArenaPro evaluation", - "tags": ["benchmark", "entropic", "crmarena"], - } - ], - } - ) - - async def handle_jsonrpc(request: Request) -> JSONResponse: - body = await request.json() - if body.get("jsonrpc") != "2.0" or "id" not in body: - return JSONResponse( - { - "jsonrpc": "2.0", - "id": None, - "error": {"code": -32600, "message": "Invalid Request"}, - } - ) - - method = body.get("method") - if method != "message/send": - return JSONResponse( - { - "jsonrpc": "2.0", - "id": body["id"], - "error": {"code": -32601, "message": f"Method not found: {method}"}, - } - ) - - loop = asyncio.get_running_loop() - result = await loop.run_in_executor(None, agent.handle_message_sync, body) - - return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result}) - - return Starlette( - routes=[ - Route("/.well-known/agent.json", agent_card, methods=["GET"]), - Route("/.well-known/agent-card.json", agent_card, methods=["GET"]), - Route("/", handle_jsonrpc, methods=["POST"]), - ] - ) - - -def start_purple_server( - agent: EntropicPurpleAgent, host: str = "127.0.0.1", port: int = 0 -) -> tuple[threading.Thread, int]: - """Start the purple agent server in a background thread. Returns (thread, actual_port).""" - import socket - import time - - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.bind((host, port)) - actual_port = sock.getsockname()[1] - - app = create_app(agent, actual_port) - config = uvicorn.Config(app, host=host, port=actual_port, log_level="warning") - server = uvicorn.Server(config) - - thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True) - thread.start() - - import httpx - - for _ in range(50): - try: - r = httpx.get(f"http://{host}:{actual_port}/.well-known/agent-card.json", timeout=0.5) - if r.status_code == 200: - break - except httpx.ConnectError: - time.sleep(0.1) - else: - log.warning("Entropic purple server did not become ready within 5s") - - return thread, actual_port diff --git a/clawloop/environments/_purple_base.py b/clawloop/environments/_purple_base.py new file mode 100644 index 00000000..7e7aeee7 --- /dev/null +++ b/clawloop/environments/_purple_base.py @@ -0,0 +1,303 @@ +# clawloop/environments/_purple_base.py +"""Shared scaffolding for A2A purple-agent servers (CAR, Entropic, ...). + +Subclasses override two bench-specific seams: +``_build_initial_messages`` (parse first-turn payload) and +``_format_a2a_response`` (wire envelope the green agent expects). +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import socket +import threading +import time +from typing import Any, ClassVar + +import httpx +import litellm +import uvicorn +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route + +from clawloop.learning_layers.harness import Harness + +log = logging.getLogger(__name__) + + +class _PurpleAgentBase: + """Base class for A2A-compliant purple agents with harness injection.""" + + # Subclasses override: + default_bench: ClassVar[str] = "" + agent_card_name: ClassVar[str] = "clawloop-purple-agent" + agent_card_description: ClassVar[str] = "ClawLoop purple agent" + agent_card_skills: ClassVar[list[dict]] = [] + + def __init__( + self, + model: str, + harness: Harness, + bench: str | None = None, + api_base: str | None = None, + api_key: str | None = None, + ): + self.model = model + self.harness = harness + self.bench = bench if bench is not None else self.default_bench + self.api_base = api_base + self.api_key = api_key + self._sessions: dict[str, list[dict]] = {} + self._tool_cache: dict[str, list[dict]] = {} + + def update_harness(self, harness: Harness) -> None: + self.harness = harness + + def clear_all_sessions(self) -> None: + self._sessions.clear() + self._tool_cache.clear() + + @staticmethod + def _convert_tools_to_openai(raw_tools: list[dict]) -> list[dict]: + """Normalize tool schemas to OpenAI function-calling format. + + Green may send tools already OpenAI-wrapped or flat; handle both. + """ + result = [] + for t in raw_tools: + if t.get("type") == "function" and "function" in t: + result.append(t) + else: + result.append( + { + "type": "function", + "function": { + "name": t["name"], + "description": t.get("description", ""), + "parameters": t.get("parameters", {}), + }, + } + ) + return result + + @staticmethod + def _normalize_assistant_msg(litellm_msg: Any) -> dict: + """Normalize litellm response to stable internal dict.""" + normalized: dict[str, Any] = { + "role": "assistant", + "content": litellm_msg.content or "", + } + if litellm_msg.tool_calls: + normalized["tool_calls"] = [ + { + "id": tc.id, + "type": "function", + "function": { + "name": tc.function.name, + "arguments": tc.function.arguments, + }, + } + for tc in litellm_msg.tool_calls + ] + return normalized + + @staticmethod + def _reconcile_tool_call_id(messages: list[dict], tool_name: str, green_id: str) -> None: + """Rewrite last assistant tool_call id to match green's id. + + The LLM needs matching ids between assistant tool_calls and tool-role + messages. Handles duplicate tool names by only rewriting calls that + still hold their LLM-generated id (not a green id already in use). + + Only the most recent assistant message is searched: tool results + always follow their triggering assistant call in this protocol, so + reconciling against older assistant messages would be incorrect. + """ + used_green_ids = { + m["tool_call_id"] for m in messages if m.get("role") == "tool" and "tool_call_id" in m + } + for msg in reversed(messages): + if msg.get("role") != "assistant" or "tool_calls" not in msg: + continue + for tc in msg["tool_calls"]: + if tc["function"]["name"] == tool_name and tc["id"] not in used_green_ids: + tc["id"] = green_id + return + return # most recent assistant msg had no matching tool_call — stop + + def _build_message_parts(self, assistant_msg: Any) -> list[dict]: + """Build the A2A ``parts`` list from an assistant message.""" + parts: list[dict] = [{"kind": "text", "text": assistant_msg.content or ""}] + if assistant_msg.tool_calls: + tool_calls = [] + for tc in assistant_msg.tool_calls: + args = tc.function.arguments + if args is None: + args = {} + elif isinstance(args, str): + try: + args = json.loads(args) + except json.JSONDecodeError: + log.warning("Malformed tool args for %s", tc.function.name) + args = {"raw": args} + tool_calls.append({"tool_name": tc.function.name, "arguments": args}) + parts.append({"kind": "data", "data": {"tool_calls": tool_calls}}) + return parts + + def _build_initial_messages(self, text_parts: list[str]) -> list[dict]: + raise NotImplementedError + + def _format_a2a_response(self, assistant_msg: Any) -> dict: + raise NotImplementedError + + def _capture_assistant(self, context_id: str, normalized: dict) -> None: + pass + + def handle_message_sync(self, jsonrpc_request: dict) -> dict: + """Handle one ``message/send`` request (sync — litellm is sync).""" + params = jsonrpc_request["params"] + msg = params["message"] + context_id = params.get("contextId", "default") + + text_parts = [p["text"] for p in msg["parts"] if p.get("kind") == "text"] + data_parts = [p["data"] for p in msg["parts"] if p.get("kind") == "data"] + + if context_id not in self._sessions: + self._sessions[context_id] = [] + messages = self._sessions[context_id] + + if not messages: + messages.extend(self._build_initial_messages(text_parts)) + for d in data_parts: + if "tools" in d: + self._tool_cache[context_id] = self._convert_tools_to_openai(d["tools"]) + else: + for d in data_parts: + if "tool_results" in d: + for tr in d["tool_results"]: + green_id = tr["tool_call_id"] + tool_name = tr.get("tool_name", "") + self._reconcile_tool_call_id(messages, tool_name, green_id) + messages.append( + { + "role": "tool", + "tool_call_id": green_id, + "content": tr["content"], + } + ) + for text in text_parts: + if text.strip(): + messages.append({"role": "user", "content": text}) + + tools = self._tool_cache.get(context_id) + completion_kwargs: dict[str, Any] = { + "model": self.model, + "messages": messages, + "temperature": 0.0, + } + if tools: + completion_kwargs["tools"] = tools + if self.api_base: + completion_kwargs["api_base"] = self.api_base + if self.api_key: + completion_kwargs["api_key"] = self.api_key + + response = litellm.completion(**completion_kwargs) + assistant_msg = response.choices[0].message + + normalized = self._normalize_assistant_msg(assistant_msg) + messages.append(normalized) + self._capture_assistant(context_id, normalized) + + return self._format_a2a_response(assistant_msg) + + +def create_app(agent: _PurpleAgentBase, port: int = 0) -> Starlette: + """Build the A2A Starlette app for a purple agent.""" + + async def agent_card(request: Request) -> JSONResponse: + return JSONResponse( + { + "name": agent.agent_card_name, + "description": agent.agent_card_description, + "url": f"http://127.0.0.1:{port}/", + "version": "0.1.0", + "protocol_version": "0.3.0", + "preferred_transport": "JSONRPC", + "default_input_modes": ["text/plain"], + "default_output_modes": ["text/plain"], + "capabilities": {"streaming": False, "push_notifications": False}, + "skills": agent.agent_card_skills, + } + ) + + async def handle_jsonrpc(request: Request) -> JSONResponse: + body = await request.json() + if body.get("jsonrpc") != "2.0" or "id" not in body: + return JSONResponse( + { + "jsonrpc": "2.0", + "id": None, + "error": {"code": -32600, "message": "Invalid Request"}, + } + ) + + method = body.get("method") + if method != "message/send": + return JSONResponse( + { + "jsonrpc": "2.0", + "id": body["id"], + "error": {"code": -32601, "message": f"Method not found: {method}"}, + } + ) + + # Run sync litellm call in thread to avoid blocking event loop + loop = asyncio.get_running_loop() + result = await loop.run_in_executor(None, agent.handle_message_sync, body) + + return JSONResponse({"jsonrpc": "2.0", "id": body["id"], "result": result}) + + return Starlette( + routes=[ + Route("/.well-known/agent.json", agent_card, methods=["GET"]), + Route("/.well-known/agent-card.json", agent_card, methods=["GET"]), + Route("/", handle_jsonrpc, methods=["POST"]), + ] + ) + + +def start_purple_server( + agent: _PurpleAgentBase, host: str = "127.0.0.1", port: int = 0 +) -> tuple[threading.Thread, int]: + """Start the purple agent server in a background thread. Returns (thread, actual_port).""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((host, port)) + actual_port = sock.getsockname()[1] + + app = create_app(agent, actual_port) + config = uvicorn.Config(app, host=host, port=actual_port, log_level="warning") + server = uvicorn.Server(config) + + thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True) + thread.start() + + for _ in range(50): + try: + r = httpx.get( + f"http://{host}:{actual_port}/.well-known/agent-card.json", + timeout=0.5, + ) + if r.status_code == 200: + break + except httpx.ConnectError: + time.sleep(0.1) + else: + log.warning("Purple server did not become ready within 5s") + + return thread, actual_port