diff --git a/agent-zero-zulip/src/agent_zero_zulip/a2a_server.py b/agent-zero-zulip/src/agent_zero_zulip/a2a_server.py new file mode 100644 index 0000000..f5a48bb --- /dev/null +++ b/agent-zero-zulip/src/agent_zero_zulip/a2a_server.py @@ -0,0 +1,94 @@ +"""A2A Agent for kagentz — processes messages via Agent Zero core directly.""" +import sys, os, asyncio, json, uuid, time + +sys.path.insert(0, "/a0") +os.environ["A2A_TOKEN"] = "8zNgdOEXzYxjQvTl" + +import uvicorn +from fastapi import FastAPI, Request +from agent import AgentContext, UserMessage, AgentContextType + +app = FastAPI() +AGENT_TOKEN = os.environ["A2A_TOKEN"] + +@app.get("/.well-known/agent.json") +async def agent_card(): + return { + "name": "kagentz", + "version": "1.0.0", + "description": "Agent Zero instance managed via Zulip", + "capabilities": {"a2a": {"version": "1.0", "authentication": ["bearer"]}}, + "skills": [{"id": "chat", "name": "Chat", "description": "General conversation via Zulip"}], + } + +@app.post("/a2a") +async def a2a_endpoint(request: Request): + body = await request.json() + method = body.get("method", "") + + if method == "tasks/send": + params = body.get("params", {}) + msg = params.get("message", {}) + parts = msg.get("parts", []) + text = " ".join(p.get("text", "") for p in parts if isinstance(p, dict)) + + task_id = f"az-{uuid.uuid4().hex[:12]}" + asyncio.create_task(process_via_agent_zero(task_id, text)) + + return {"jsonrpc": "2.0", "result": {"id": task_id, "status": {"state": "working"}}} + + elif method == "tasks/get": + params = body.get("params", {}) + task_id = params.get("id", "") + status, result = task_results.get(task_id, ("pending", None)) + + resp = {"jsonrpc": "2.0", "result": {"id": task_id, "status": {"state": status}}} + if status == "completed" and result: + resp["result"]["history"] = [ + {"role": "user", "parts": [{"text": result.get("input", "")}]}, + {"role": "assistant", "parts": [{"text": result.get("output", "")}]}, + ] + elif status == "failed": + resp["result"]["status"]["message"] = {"text": str(result or "Unknown error")} + + return resp + + return {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}} + +task_results = {} + +async def process_via_agent_zero(task_id: str, message: str): + """Process a message through Agent Zero's core.""" + try: + # Create a new context and send the message + ctx = AgentContext.get(uuid.uuid4().hex) + + # Run in executor to avoid blocking the event loop + loop = asyncio.get_event_loop() + + def _run(): + try: + # Send user message and get response + msg = UserMessage(text=message, context=ctx) + result = ctx.agent.handle_message(msg) + + # Extract text response + if hasattr(result, 'content'): + output = result.content + elif isinstance(result, str): + output = result + else: + output = str(result) + + return output + except Exception as e: + return f"Error: {e}" + + output = await loop.run_in_executor(None, _run) + task_results[task_id] = ("completed", {"input": message, "output": output}) + + except Exception as e: + task_results[task_id] = ("failed", str(e)) + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8001, log_level="info")