Skip to content

C
clawphone
Commit b46ac71f authored by Lisa (AI Assistant)'s avatar Lisa (AI Assistant)
Browse files
Options

Centralized architecture: single MCP server, HTTPS webhooks, register_agent tool

parent 36575762
Pipeline #272 canceled with stages
      Hide whitespace changes
      Inline Side-by-side
      Showing with 285 additions and 162 deletions
      README.md
      View file @ b46ac71f
      # ClawPhone
      A job queue MCP server for OpenClaw agents. Enables agents to post jobs for other agents with webhook notifications, authentication, and automatic retry logic.
      A **centralized** job queue MCP server for OpenClaw agents. Enables multiple isolated OpenClaw instances to communicate and delegate jobs to each other.
      ## Architecture
      ```
      ┌─────────────┐ ┌─────────────────┐ ┌─────────────┐
      │ OpenClaw │ │ MCP Server │ │ OpenClaw │
      │ Lisa ─────┼────►│ (Central) │◄────┼─ Nimpho │
      │ │ MCP │ │◄────┤ │
      └─────────────┘ └────────┬────────┘ └─────────────┘
      │
      Webhook (HTTPS)
      ```
      - **One central MCP server** (can run on any server - one of the instances or a dedicated box)
      - **All OpenClaw instances** connect to this MCP server via MCP
      - **Jobs** are posted from any instance to any other instance via MCP
      - **Notifications** sent via HTTPS webhooks with `verify=False`
      ## Features
      - **Job Queue**: Post, claim, and track jobs between agents
      - **Per-Agent Webhooks**: Each agent has its own webhook URL for notifications
      - **Bearer Token Authentication**: Secure communication between MCP server and agents
      - **Auto-Retry**: Jobs that aren't claimed get retried at increasing intervals (1min → 2min → 5min → 10min)
      - **HTTPS**: Self-signed certificate support
      - **SQLite**: Simple persistent storage
      - **Central Job Queue**: Single MCP server handles jobs from all instances
      - **Agent Registry**: Each OpenClaw instance registers with its webhook URL
      - **HTTPS Webhooks**: Server notifies agents via HTTPS (verify=false)
      - **Bearer Token Auth**: Secure communication
      - **Auto-Retry**: Unclaimed jobs retry at intervals (1→2→5→10 min)
      - **SQLite Storage**: Simple persistent queue
      ## Installation
      ```bash
      # Clone or copy to target directory
      cp -r clawphone /home/share/clawphone
      # Clone
      git clone https://git.nexlab.net/lisa/clawphone.git /opt/clawphone
      # Install dependencies (if needed)
      # Install deps
      pip install fastapi uvicorn aiosqlite httpx cryptography
      # Copy init script
      sudo cp etc/init.d/clawphone /etc/init.d/
      sudo chmod +x /etc/init.d/clawphone
      sudo update-rc.d clawphone defaults
      # Configure agents
      cp agents.json.example agents.json
      # Edit agents.json with your instance URLs and tokens
      ```
      ## Configuration
      Edit `/home/share/clawphone/agents.json`:
      Create `agents.json`:
      ```json
      {
      "agent_name": {
      "hook": "https://your-server.com/api/agent/hook",
      "token": "agent_bearer_token"
      "lisa": {
      "hook": "https://lisa.nexlab.net/hooks/agent",
      "token": "your-openclaw-hook-token"
      },
      "nimpho": {
      "hook": "https://nimpho.nexlab.net/hooks/agent",
      "token": "your-openclaw-hook-token"
      },
      "postino": {
      "hook": "https://postino.nexlab.net/hooks/agent",
      "token": "your-openclaw-hook-token"
      }
      }
      ```
      Optionally set a global API token in `/home/share/clawphone/.env`:
      Optionally set a global API token in `.env`:
      ```
      CLAWPHONE_TOKEN=your_global_token
      ```
      ## Usage
      ### Start Server
      ## Running
      ```bash
      # Manual
      sudo /etc/init.d/clawphone start
      python3 /opt/clawphone/mcp_server.py
      # Or directly
      python3 /home/share/clawphone/mcp_server.py
      # Or with the init script
      sudo /etc/init.d/clawphone start
      ```
      ### MCP Tools
      ## MCP Tools
      | Tool | Description |
      |------|-------------|
      | `post_job` | Post a new job for an agent |
      | `register_agent` | Register an agent (name, hook URL, token) |
      | `post_job` | Post a job for a specific agent |
      | `claim_job` | Claim a pending job |
      | `update_job_status` | Update job to "working" or "done" |
      | `list_jobs` | List jobs (filter by agent/status) |
      | `get_job` | Get job details |
      | `list_agents` | List configured agents |
      | `list_agents` | List registered agents |
      ### API Examples
      ## API Examples
      ```bash
      # List jobs
      curl -k -H "Authorization: Bearer TOKEN" \
      https://localhost:8765/tools/list_jobs
      # Register an agent
      curl -k -X POST https://localhost:8765/tools/register_agent \
      -H "Authorization: Bearer TOKEN" \
      -H "Content-Type: application/json" \
      -d '{"name":"lisa","hook":"https://lisa.nexlab.net/hooks/agent","token":"hook-token"}'
      # Post a job
      curl -k -X POST -H "Authorization: Bearer TOKEN" \
      # Post a job (Lisa asking Nimpho to do something)
      curl -k -X POST https://localhost:8765/tools/post_job \
      -H "Authorization: Bearer TOKEN" \
      -H "Content-Type: application/json" \
      -d '{"sender":"agent-a","target_agent":"agent-b","title":"Do something"}' \
      https://localhost:8765/tools/post_job
      -d '{"sender":"lisa","target_agent":"nimpho","title":"Check server status","description":"Run uptime command on ganeti1"}'
      # Claim a job
      curl -k -X POST -H "Authorization: Bearer TOKEN" \
      # Claim a job (Nimpho claims the job)
      curl -k -X POST https://localhost:8765/tools/claim_job \
      -H "Authorization: Bearer TOKEN" \
      -H "Content-Type: application/json" \
      -d '{"job_id":"uuid-here","agent":"agent-b"}' \
      https://localhost:8765/tools/claim_job
      -d '{"job_id":"uuid-here","agent":"nimpho"}'
      # Update job status
      curl -k -X POST -H "Authorization: Bearer TOKEN" \
      curl -k -X POST https://localhost:8765/tools/update_job_status \
      -H "Authorization: Bearer TOKEN" \
      -H "Content-Type: application/json" \
      -d '{"job_id":"uuid-here","status":"done","agent":"agent-b"}' \
      https://localhost:8765/tools/update_job_status
      -d '{"job_id":"uuid-here","status":"done","agent":"nimpho","result":"Server is up"}'
      ```
      ## Webhook Events
      The MCP server sends POST requests to agent webhooks with these events:
      The MCP server sends POST requests to agent webhooks:
      | Event | Description |
      |-------|-------------|
      | Event | Payload |
      |-------|---------|
      | `new_job` | New job posted for agent |
      | `job_pending` | Job retry notification |
      | `job_claimed` | Job was claimed |
      | `job_status_changed` | Job status updated |
      | `job_failed` | Job expired (not claimed) |
      | `job_done` | Job completed with result |
      | `job_failed` | Job expired |
      All webhooks use HTTPS with `verify=False`.
      ## License
      ......@@ -116,8 +143,6 @@ Stefy Lanza <stefy@nexlab.net>
      ## Donations
      If you find this project useful, consider donating:
      | Crypto | Address |
      |--------|---------|
      | Bitcoin (BTC) | `bc1q3zlkpu95amtcltsk85y0eacyzzk29v68tgc5hx` |
      ......@@ -126,6 +151,4 @@ If you find this project useful, consider donating:
      ---
      <p align="center">
      Made with 🧠 and 🔧 by Stefy
      </p>
      <p align="center">Made with 🧠 and 🔧 by Stefy</p>
      agents.json
      View file @ b46ac71f
      {
      "coderino": {
      "hook": "https://lisa.nexlab.net/hooks/agent",
      "token": "clawphone_hook_token_2024"
      },
      "marketta": {
      "hook": "https://lisa.nexlab.net/hooks/agent",
      "token": "clawphone_hook_token_2024"
      },
      "postino": {
      "hook": "https://lisa.nexlab.net/hooks/agent",
      "token": "clawphone_hook_token_2024"
      "example": {
      "hook": "https://your-instance.nexlab.net/hooks/agent",
      "token": "your-openclaw-hook-token"
      }
      }
      mcp_server.py
      View file @ b46ac71f
      #!/usr/bin/env python3
      """
      ClawPhone MCP Server - Job Queue for Agents
      With per-agent Bearer token authentication
      ClawPhone - Centralized Job Queue MCP Server for OpenClaw Agents
      All webhooks use HTTPS with verify=False
      """
      import os
      ......@@ -12,6 +12,7 @@ import sqlite3
      from datetime import datetime, timedelta
      from pathlib import Path
      from contextlib import asynccontextmanager
      from typing import Optional
      from fastapi import FastAPI, HTTPException, Header
      import uvicorn
      ......@@ -30,34 +31,62 @@ HOST = os.getenv("CLAWPHONE_HOST", "0.0.0.0")
      CERT_PATH = os.getenv("CLAWPHONE_CERT", "/home/share/clawphone/server.crt")
      KEY_PATH = os.getenv("CLAWPHONE_KEY", "/home/share/clawphone/server.key")
      API_TOKEN = os.getenv("CLAWPHONE_TOKEN", "")
      AGENTS_CONFIG = os.getenv("CLAWPHONE_AGENTS", "") # JSON: {"agent1": {"hook": "url", "token": "xxx"}, ...}
      app = FastAPI(title="ClawPhone Job Queue")
      HTTP_CLIENT = httpx.AsyncClient(verify=False, timeout=30.0)
      HTTP_CLIENT = httpx.AsyncClient(verify=False, timeout=30.0) # HTTPS verify=False
      RETRY_INTERVALS = [1, 2, 5, 10]
      # Parse agents config
      # In-memory agent registry (loaded from DB)
      AGENTS = {}
      if AGENTS_CONFIG:
      try:
      AGENTS = json.loads(AGENTS_CONFIG)
      except:
      pass
      # Also check for individual env vars (CLAWPHONE_AGENT_<name>_HOOK, CLAWPHONE_AGENT_<name>_TOKEN)
      for key, value in os.environ.items():
      if key.startswith("CLAWPHONE_AGENT_") and key.endswith("_HOOK"):
      agent_name = key[16:-5] # Remove CLAWPHONE_AGENT_ and _HOOK
      token_key = f"CLAWPHONE_AGENT_{agent_name}_TOKEN"
      AGENTS[agent_name] = {"hook": value, "token": os.environ.get(token_key, "")}
      async def init_db():
      """Initialize database"""
      async with aiosqlite.connect(DB_PATH) as db:
      # Agents table
      await db.execute("""
      CREATE TABLE IF NOT EXISTS agents (
      name TEXT PRIMARY KEY,
      hook TEXT NOT NULL,
      token TEXT,
      registered_at TEXT NOT NULL
      )
      """)
      # Jobs table
      await db.execute("""
      CREATE TABLE IF NOT EXISTS jobs (
      id TEXT PRIMARY KEY,
      sender TEXT NOT NULL,
      target_agent TEXT NOT NULL,
      title TEXT NOT NULL,
      description TEXT,
      status TEXT DEFAULT 'pending',
      result TEXT,
      created_at TEXT NOT NULL,
      updated_at TEXT NOT NULL,
      retry_count INTEGER DEFAULT 0,
      next_retry_at TEXT
      )
      """)
      await db.commit()
      # Load agents into memory
      await load_agents()
      print(f"Loaded {len(AGENTS)} agents: {list(AGENTS.keys())}")
      async def load_agents():
      """Load agents from DB"""
      global AGENTS
      async with aiosqlite.connect(DB_PATH) as db:
      cur = await db.execute("SELECT name, hook, token FROM agents")
      rows = await cur.fetchall()
      AGENTS = {r[0]: {"hook": r[1], "token": r[2]} for r in rows}
      print(f"Loaded {len(AGENTS)} agents: {list(AGENTS.keys())}")
      def verify_token(authorization: str = Header(None), agent: str = None) -> str:
      def verify_token(authorization: str = Header(None)) -> bool:
      """Verify Bearer token"""
      if not API_TOKEN and not AGENTS:
      return "anonymous"
      if not API_TOKEN:
      return True # No auth configured
      if not authorization:
      raise HTTPException(status_code=401, detail="Missing Authorization header")
      ......@@ -66,37 +95,19 @@ def verify_token(authorization: str = Header(None), agent: str = None) -> str:
      raise HTTPException(status_code=401, detail="Invalid authorization format")
      token = authorization[7:]
      if token != API_TOKEN:
      raise HTTPException(status_code=401, detail="Invalid token")
      # Check global token
      if API_TOKEN and token == API_TOKEN:
      return "global"
      # Check agent-specific token
      if agent and agent in AGENTS and AGENTS[agent].get("token"):
      if token == AGENTS[agent]["token"]:
      return agent
      raise HTTPException(status_code=401, detail="Invalid token")
      async def init_db():
      async with aiosqlite.connect(DB_PATH) as db:
      await db.execute("""
      CREATE TABLE IF NOT EXISTS jobs (
      id TEXT PRIMARY KEY, sender TEXT NOT NULL, target_agent TEXT NOT NULL,
      title TEXT NOT NULL, description TEXT, status TEXT DEFAULT 'pending',
      created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
      retry_count INTEGER DEFAULT 0, next_retry_at TEXT, hook_url TEXT, hook_token TEXT
      )
      """)
      await db.commit()
      return True
      def generate_self_signed_cert():
      """Generate self-signed HTTPS cert"""
      cert_file = Path(CERT_PATH)
      key_file = Path(KEY_PATH)
      if cert_file.exists() and key_file.exists():
      return
      key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
      subject = issuer = x509.Name([
      x509.NameAttribute(NameOID.COUNTRY_NAME, "ZZ"),
      ......@@ -109,52 +120,72 @@ def generate_self_signed_cert():
      cert = x509.CertificateBuilder().subject_name(subject).issuer_name(issuer).public_key(
      key.public_key()).serial_number(x509.random_serial_number()).not_valid_before(
      now).not_valid_after(now + timedelta(days=365)).sign(key, hashes.SHA256(), default_backend())
      key_file.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()))
      cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
      async def notify_hook(hook_url: str, hook_token: str, data: dict) -> bool:
      async def notify_hook(agent_name: str, event: str, data: dict) -> bool:
      """Send webhook to agent (HTTPS, verify=False)"""
      if agent_name not in AGENTS:
      print(f"Agent {agent_name} not found")
      return False
      agent = AGENTS[agent_name]
      hook_url = agent.get("hook")
      hook_token = agent.get("token")
      if not hook_url:
      print("No hook_url configured")
      print(f"No hook URL for agent {agent_name}")
      return False
      headers = {}
      if hook_token:
      headers["Authorization"] = f"Bearer {hook_token}"
      payload = {
      "event": event,
      "agent": agent_name,
      "timestamp": datetime.now().isoformat(),
      **data
      }
      try:
      r = await HTTP_CLIENT.post(hook_url, json=data, headers=headers)
      print(f"Hook to {data.get('target_agent')}: {r.status_code} - {data.get('event')}")
      return r.status_code == 200
      r = await HTTP_CLIENT.post(hook_url, json=payload, headers=headers)
      print(f"Webhook to {agent_name} ({event}): {r.status_code}")
      return r.status_code in (200, 201)
      except Exception as e:
      print(f"Hook failed: {e}")
      print(f"Webhook failed: {e}")
      return False
      async def check_pending_jobs():
      """Background task to retry pending jobs"""
      while True:
      try:
      async with aiosqlite.connect(DB_PATH) as db:
      now = datetime.now().isoformat()
      cur = await db.execute(
      "SELECT id, sender, target_agent, title, description, hook_url, hook_token FROM jobs WHERE status='pending' AND next_retry_at IS NOT NULL AND next_retry_at <= ?",
      "SELECT id, sender, target_agent, title, description FROM jobs WHERE status='pending' AND next_retry_at IS NOT NULL AND next_retry_at <= ?",
      (now,))
      rows = await cur.fetchall()
      for row in rows:
      job_id, sender, target, title, desc, hook_url, hook_token = row
      job_id, sender, target, title, desc = row
      cur2 = await db.execute("SELECT retry_count FROM jobs WHERE id = ?", (job_id,))
      res = await cur2.fetchone()
      retry_count = res[0] if res else 0
      if retry_count >= len(RETRY_INTERVALS):
      await db.execute("UPDATE jobs SET status='failed', updated_at=? WHERE id=?", (datetime.now().isoformat(), job_id))
      await notify_hook(hook_url, hook_token, {"event": "job_failed", "job_id": job_id, "sender": sender, "target_agent": target, "title": title})
      await notify_hook(target, "job_failed", {"job_id": job_id, "sender": sender, "title": title})
      else:
      await notify_hook(hook_url, hook_token, {"event": "job_pending", "job_id": job_id, "sender": sender, "target_agent": target, "title": title, "retry_count": retry_count + 1})
      # Notify about pending retry
      await notify_hook(target, "job_pending", {"job_id": job_id, "sender": sender, "title": title, "retry": retry_count + 1})
      interval = RETRY_INTERVALS[min(retry_count, len(RETRY_INTERVALS)-1)]
      next_retry = (datetime.now() + timedelta(minutes=interval)).isoformat()
      await db.execute("UPDATE jobs SET retry_count=retry_count+1, next_retry_at=?, updated_at=? WHERE id=?", (next_retry, datetime.now().isoformat(), job_id))
      await db.execute("UPDATE jobs SET retry_count=retry_count+1, next_retry_at=?, updated_at=? WHERE id=?",
      (next_retry, datetime.now().isoformat(), job_id))
      await db.commit()
      except Exception as e:
      print(f"Background error: {e}")
      ......@@ -175,127 +206,204 @@ app = FastAPI(lifespan=lifespan)
      async def list_tools(authorization: str = Header(None)):
      verify_token(authorization)
      return {"tools": [
      {"name": "post_job", "description": "Post a new job", "inputSchema": {"type": "object", "properties": {"sender": {}, "target_agent": {}, "title": {}, "description": {}}, "required": ["sender", "target_agent", "title"]}},
      {"name": "claim_job", "description": "Claim a job", "inputSchema": {"type": "object", "properties": {"job_id": {}, "agent": {}}, "required": ["job_id", "agent"]}},
      {"name": "update_job_status", "description": "Update job status", "inputSchema": {"type": "object", "properties": {"job_id": {}, "status": {"enum": ["working", "done"]}, "agent": {}}, "required": ["job_id", "status", "agent"]}},
      {"name": "list_jobs", "description": "List jobs", "inputSchema": {"type": "object", "properties": {"agent": {}, "status": {}}}},
      {"name": "get_job", "description": "Get job", "inputSchema": {"type": "object", "properties": {"job_id": {}}, "required": ["job_id"]}},
      {"name": "list_agents", "description": "List configured agents", "inputSchema": {"type": "object", "properties": {}}}
      {"name": "register_agent", "description": "Register an agent with its webhook URL",
      "inputSchema": {"type": "object", "properties": {"name": {}, "hook": {}, "token": {}}, "required": ["name", "hook"]}},
      {"name": "post_job", "description": "Post a new job for an agent",
      "inputSchema": {"type": "object", "properties": {"sender": {}, "target_agent": {}, "title": {}, "description": {}}, "required": ["sender", "target_agent", "title"]}},
      {"name": "claim_job", "description": "Claim a pending job",
      "inputSchema": {"type": "object", "properties": {"job_id": {}, "agent": {}}, "required": ["job_id", "agent"]}},
      {"name": "update_job_status", "description": "Update job status",
      "inputSchema": {"type": "object", "properties": {"job_id": {}, "status": {"enum": ["working", "done", "failed"]}, "agent": {}, "result": {}}, "required": ["job_id", "status", "agent"]}},
      {"name": "list_jobs", "description": "List jobs",
      "inputSchema": {"type": "object", "properties": {"agent": {}, "status": {}}}},
      {"name": "get_job", "description": "Get job details",
      "inputSchema": {"type": "object", "properties": {"job_id": {}}, "required": ["job_id"]}},
      {"name": "list_agents", "description": "List registered agents",
      "inputSchema": {"type": "object", "properties": {}}}
      ]}
      @app.get("/tools/list_agents")
      async def list_agents(authorization: str = Header(None)):
      verify_token(authorization)
      # Return agents without tokens
      await load_agents() # Refresh from DB
      return {"agents": {k: {"hook": v.get("hook")} for k, v in AGENTS.items()}}
      @app.post("/tools/register_agent")
      async def tool_register_agent(data: dict, authorization: str = Header(None)):
      """Register an agent with its webhook URL"""
      verify_token(authorization)
      name = data.get("name")
      hook = data.get("hook")
      token = data.get("token", "")
      if not name or not hook:
      return {"success": False, "error": "name and hook are required"}
      # Ensure HTTPS
      if not hook.startswith("https://"):
      hook = hook.replace("http://", "https://")
      async with aiosqlite.connect(DB_PATH) as db:
      await db.execute(
      "INSERT OR REPLACE INTO agents (name, hook, token, registered_at) VALUES (?, ?, ?, ?)",
      (name, hook, token, datetime.now().isoformat()))
      await db.commit()
      await load_agents()
      return {"success": True, "agent": name, "hook": hook}
      @app.post("/tools/post_job")
      async def tool_post_job(data: dict, authorization: str = Header(None)):
      """Post a new job"""
      verify_token(authorization)
      sender = data.get("sender")
      target_agent = data.get("target_agent")
      hook_url = None
      hook_token = None
      title = data.get("title")
      description = data.get("description", "")
      # Get hook from agent config
      if target_agent and target_agent in AGENTS:
      hook_url = AGENTS[target_agent].get("hook")
      hook_token = AGENTS[target_agent].get("token")
      if not all([sender, target_agent, title]):
      return {"success": False, "error": "sender, target_agent, and title are required"}
      job_id = str(uuid.uuid4())
      now = datetime.now().isoformat()
      first_retry = (datetime.now() + timedelta(minutes=RETRY_INTERVALS[0])).isoformat()
      async with aiosqlite.connect(DB_PATH) as db:
      await db.execute("INSERT INTO jobs VALUES (?, ?, ?, ?, ?, 'pending', ?, ?, 0, ?, ?, ?)",
      (job_id, data["sender"], target_agent, data["title"], data.get("description",""), now, now, first_retry, hook_url, hook_token))
      await db.execute(
      "INSERT INTO jobs VALUES (?, ?, ?, ?, ?, 'pending', NULL, ?, ?, 0, ?)",
      (job_id, sender, target_agent, title, description, now, now, first_retry))
      await db.commit()
      # Notify target agent
      if hook_url:
      await notify_hook(hook_url, hook_token, {"event": "new_job", "job_id": job_id, "sender": data["sender"], "target_agent": target_agent, "title": data["title"], "description": data.get("description","")})
      # Notify target agent via webhook (HTTPS)
      await notify_hook(target_agent, "new_job", {
      "job_id": job_id,
      "sender": sender,
      "title": title,
      "description": description
      })
      return {"success": True, "job_id": job_id}
      @app.post("/tools/claim_job")
      async def tool_claim_job(data: dict, authorization: str = Header(None)):
      """Claim a job"""
      verify_token(authorization)
      job_id, agent = data["job_id"], data["agent"]
      job_id = data.get("job_id")
      agent = data.get("agent")
      if not job_id or not agent:
      return {"success": False, "error": "job_id and agent are required"}
      async with aiosqlite.connect(DB_PATH) as db:
      cur = await db.execute("SELECT status, hook_url, hook_token, sender FROM jobs WHERE id = ?", (job_id,))
      cur = await db.execute("SELECT status, target_agent, sender FROM jobs WHERE id = ?", (job_id,))
      row = await cur.fetchone()
      if not row: return {"success": False, "error": "Not found"}
      if row[0] != "pending": return {"success": False, "error": f"Job is {row[0]}"}
      # Notify sender
      sender = row[3]
      if sender and sender in AGENTS:
      sender_hook = AGENTS.get(sender, {}).get("hook")
      sender_token = AGENTS.get(sender, {}).get("token")
      if sender_hook:
      await notify_hook(sender_hook, sender_token, {"event": "job_claimed", "job_id": job_id, "agent": agent})
      if not row:
      return {"success": False, "error": "Job not found"}
      if row[0] != "pending":
      return {"success": False, "error": f"Job is {row[0]}"}
      if row[1] != agent:
      return {"success": False, "error": "Job is for a different agent"}
      await db.execute("UPDATE jobs SET status='claimed', updated_at=?, next_retry_at=NULL WHERE id=?", (datetime.now().isoformat(), job_id))
      await db.execute("UPDATE jobs SET status='claimed', updated_at=?, next_retry_at=NULL WHERE id=?",
      (datetime.now().isoformat(), job_id))
      await db.commit()
      # Notify sender
      sender = row[2]
      if sender:
      await notify_hook(sender, "job_claimed", {"job_id": job_id, "agent": agent, "title": "Job claimed"})
      return {"success": True}
      @app.post("/tools/update_job_status")
      async def tool_update_job_status(data: dict, authorization: str = Header(None)):
      """Update job status"""
      verify_token(authorization)
      job_id, status, agent = data["job_id"], data["status"], data["agent"]
      job_id = data.get("job_id")
      status = data.get("status")
      agent = data.get("agent")
      result = data.get("result", "")
      if not all([job_id, status, agent]):
      return {"success": False, "error": "job_id, status, and agent are required"}
      async with aiosqlite.connect(DB_PATH) as db:
      cur = await db.execute("SELECT status, hook_url, hook_token, sender FROM jobs WHERE id = ?", (job_id,))
      cur = await db.execute("SELECT status, target_agent, sender FROM jobs WHERE id = ?", (job_id,))
      row = await cur.fetchone()
      if not row: return {"success": False, "error": "Not found"}
      # Notify sender
      sender = row[3]
      if sender and sender in AGENTS:
      sender_hook = AGENTS.get(sender, {}).get("hook")
      sender_token = AGENTS.get(sender, {}).get("token")
      if sender_hook:
      await notify_hook(sender_hook, sender_token, {"event": "job_status_changed", "job_id": job_id, "old_status": row[0], "new_status": status, "agent": agent})
      if not row:
      return {"success": False, "error": "Job not found"}
      await db.execute("UPDATE jobs SET status=?, updated_at=? WHERE id=?", (status, datetime.now().isoformat(), job_id))
      if row[1] != agent:
      return {"success": False, "error": "Job is for a different agent"}
      await db.execute("UPDATE jobs SET status=?, result=?, updated_at=? WHERE id=?",
      (status, result, datetime.now().isoformat(), job_id))
      await db.commit()
      # Notify sender
      sender = row[2]
      if sender:
      await notify_hook(sender, "job_status_changed", {
      "job_id": job_id,
      "status": status,
      "result": result
      })
      return {"success": True}
      @app.get("/tools/list_jobs")
      async def tool_list_jobs(agent: str = None, status: str = None, authorization: str = Header(None)):
      """List jobs"""
      verify_token(authorization)
      q, p = "SELECT * FROM jobs WHERE 1=1", []
      if agent: q += " AND (target_agent=? OR sender=?)"; p += [agent, agent]
      if status: q += " AND status=?"; p += [status]
      if agent:
      q += " AND (target_agent=? OR sender=?)"
      p += [agent, agent]
      if status:
      q += " AND status=?"
      p += [status]
      q += " ORDER BY created_at DESC"
      async with aiosqlite.connect(DB_PATH) as db:
      db.row_factory = sqlite3.Row
      cur = await db.execute(q, p)
      rows = await cur.fetchall()
      return {"jobs": [dict(r) for r in rows]}
      @app.get("/tools/get_job")
      async def tool_get_job(job_id: str, authorization: str = Header(None)):
      """Get job details"""
      verify_token(authorization)
      async with aiosqlite.connect(DB_PATH) as db:
      db.row_factory = sqlite3.Row
      cur = await db.execute("SELECT * FROM jobs WHERE id = ?", (job_id,))
      row = await cur.fetchone()
      return {"job": dict(row)} if row else {"error": "Not found"}
      @app.get("/health")
      async def health():
      return {"status": "ok", "agents": len(AGENTS), "auth": bool(API_TOKEN)}
      return {"status": "ok", "agents": len(AGENTS)}
      if __name__ == "__main__":
      ......
      Markdown is supported
      0% or
      You are about to add 0 people to the discussion. Proceed with caution.
      Finish editing this message first!
      Please register or to comment