Spaces:
Sleeping
Sleeping
| import json | |
| import subprocess | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any, Dict | |
| from uuid import uuid4 | |
| import gradio as gr | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| import uvicorn | |
| ROOT = Path(__file__).resolve().parent | |
| META_PATH = ROOT / "agent.json" | |
| def load_meta() -> Dict[str, Any]: | |
| try: | |
| return json.loads(META_PATH.read_text(encoding="utf-8")) | |
| except Exception: | |
| return {"id": "sin-code-base", "name": "A2A SIN-Code Base", "version": "1.0.0"} | |
| def build_card(base_url: str) -> Dict[str, Any]: | |
| meta = load_meta() | |
| meta = dict(meta) | |
| meta["endpoints"] = dict(meta.get("endpoints") or {}) | |
| meta["endpoints"]["gradio_ui"] = base_url | |
| meta["endpoints"]["a2a_api"] = f"{base_url}/a2a/v1" | |
| meta["endpoints"]["agent_card"] = f"{base_url}/.well-known/agent-card.json" | |
| return meta | |
| status = {"running": False, "last_task": None} | |
| def call_opencode(prompt: str, timeout: int = 300) -> str: | |
| try: | |
| result = subprocess.run( | |
| ["opencode", "run", prompt, "--format", "json"], | |
| capture_output=True, | |
| text=True, | |
| timeout=timeout, | |
| ) | |
| parts = [] | |
| for line in result.stdout.splitlines(): | |
| try: | |
| ev = json.loads(line) | |
| if ev.get("type") == "text": | |
| parts.append(ev.get("part", {}).get("text", "")) | |
| except json.JSONDecodeError: | |
| pass | |
| return "".join(parts).strip() | |
| except subprocess.TimeoutExpired: | |
| return "Error: opencode call timed out" | |
| except Exception as e: | |
| return f"Error: {str(e)}" | |
| app = FastAPI() | |
| async def health() -> Dict[str, Any]: | |
| meta = load_meta() | |
| return { | |
| "ok": True, | |
| "agent": meta.get("id"), | |
| "running": status["running"], | |
| "version": meta.get("version"), | |
| } | |
| async def root() -> str: | |
| return "<html><body><h1>A2A SIN-Code Agent</h1><p>See <a href='/ui'>UI</a></p></body></html>" | |
| async def agent_card(request: Request) -> JSONResponse: | |
| base_url = f"{'https' if request.url.scheme == 'https' else 'http'}://{request.headers.get('host', '')}" | |
| return JSONResponse(build_card(base_url)) | |
| async def a2a(request: Request) -> JSONResponse: | |
| rpc = await request.json() | |
| rpc_id = rpc.get("id") | |
| method = rpc.get("method") | |
| if rpc.get("jsonrpc") != "2.0" or not isinstance(method, str): | |
| return JSONResponse( | |
| { | |
| "jsonrpc": "2.0", | |
| "id": rpc_id, | |
| "error": {"code": -32600, "message": "invalid_request"}, | |
| }, | |
| status_code=400, | |
| ) | |
| if method == "agent/getCard": | |
| base_url = f"{'https' if request.url.scheme == 'https' else 'http'}://{request.headers.get('host', '')}" | |
| return JSONResponse({"jsonrpc": "2.0", "id": rpc_id, "result": build_card(base_url)}) | |
| if method == "message/send": | |
| params = rpc.get("params", {}) | |
| message = params.get("message", {}) | |
| parts = message.get("parts", []) | |
| data = None | |
| for part in parts: | |
| if part.get("type") == "data": | |
| data = part.get("data", {}) | |
| break | |
| action = data.get("action", "agent.help") if isinstance(data, dict) else "agent.help" | |
| if action == "agent.help": | |
| result = {"help": "Specialized coding agent", "skills": load_meta().get("skills", [])} | |
| elif action.startswith("sin.code."): | |
| prompt = f"You are {load_meta().get('name')}. {data.get('description', '')}" | |
| generated = call_opencode(prompt) | |
| result = {"generated_code": generated, "action": action, "status": "completed"} | |
| else: | |
| result = {"error": f"Unknown action: {action}"} | |
| task = { | |
| "id": str(uuid4()), | |
| "kind": "task", | |
| "status": { | |
| "state": "completed", | |
| "timestamp": datetime.utcnow().isoformat() + "Z", | |
| "message": {"role": "agent", "parts": [{"type": "text", "text": "completed"}]}, | |
| }, | |
| "artifacts": [ | |
| { | |
| "id": str(uuid4()), | |
| "name": "result", | |
| "description": "Task result", | |
| "parts": [{"type": "data", "data": result}], | |
| } | |
| ], | |
| "metadata": {"action": action}, | |
| } | |
| return JSONResponse({"jsonrpc": "2.0", "id": rpc_id, "result": task}) | |
| return JSONResponse( | |
| {"jsonrpc": "2.0", "id": rpc_id, "error": {"code": -32601, "message": "method_not_found"}}, | |
| status_code=404, | |
| ) | |
| demo = gr.Blocks(title="A2A SIN-Code Agent") | |
| with demo: | |
| meta = load_meta() | |
| gr.Markdown(f"# {meta.get('name', 'A2A SIN-Code Agent')}") | |
| gr.Textbox(label="Agent ID", value=meta.get("id"), interactive=False) | |
| gr.Textbox(label="Status", value="Running", interactive=False) | |
| app = gr.mount_gradio_app(app, demo, path="/ui") | |
| if __name__ == "__main__": | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |