Skip to content

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

MCP server: Use FastMCP for HTTP transport 

- Rewrote to use FastMCP for simpler SSE/HTTP transport
- Supports stdio, SSE, and streamable HTTP modes
- HTTP mode works with mcporter on port 8765
parent 83a4b4df
Show whitespace changes
Inline Side-by-side
Showing with 136 additions and 582 deletions
mcp_server.py
View file @ 34f5a7c1
#!/home/share/clawphone/venv/bin/python3
"""
ClawPhone - Centralized Job Queue MCP Server for OpenClaw Agents
All webhooks use HTTPS with verify=False
ClawPhone - MCP Server using FastMCP for SSE transport
"""
import os
......@@ -12,564 +11,202 @@ import argparse
import secrets
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime, timedelta
from pathlib import Path
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Optional
from fastapi import FastAPI, HTTPException, Header, Body, Request
from fastapi.responses import PlainTextResponse
import uvicorn
import httpx
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend
import uuid
from mcp.server.fastmcp import FastMCP
import aiosqlite
import sqlite3
import uuid
import httpx
# Default paths
DEFAULT_DB_PATH = "/home/share/clawphone/queue.db"
DEFAULT_CERT_PATH = "/home/share/clawphone/server.crt"
DEFAULT_KEY_PATH = "/home/share/clawphone/server.key"
DEFAULT_DB_PATH = "/home/lisa/.openclaw/workspace/working/clawphone/queue.db"
DEFAULT_LOG_PATH = "/var/log/clawphone"
DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 8765
# Parse CLI arguments
# Parse CLI
parser = argparse.ArgumentParser(description="ClawPhone MCP Server")
parser.add_argument("--host", default=None, help="Host to bind to")
parser.add_argument("--port", type=int, default=None, help="Port to bind to")
parser.add_argument("--token", default=None, help="Server authentication token")
parser.add_argument("--generate-token", action="store_true", help="Generate a new token")
parser.add_argument("--db", default=None, help="Database path")
parser.add_argument("--cert", default=None, help="TLS certificate path")
parser.add_argument("--key", default=None, help="TLS key path")
parser.add_argument("--syslog", action="store_true", help="Use syslog for logging")
parser.add_argument("--log-dir", default=None, help="Log directory")
parser.add_argument("--host", default=None)
parser.add_argument("--port", type=int, default=None)
parser.add_argument("--token", default=None)
parser.add_argument("--db", default=None)
parser.add_argument("--syslog", action="store_true")
parser.add_argument("--log-dir", default=None)
parser.add_argument("--stdio", action="store_true")
parser.add_argument("--http", action="store_true")
args = parser.parse_args()
# Configuration from env vars (override with CLI args)
# Config
DB_PATH = args.db or os.getenv("CLAWPHONE_DB", DEFAULT_DB_PATH)
HOST = args.host or os.getenv("CLAWPHONE_HOST", DEFAULT_HOST)
PORT = args.port or int(os.getenv("CLAWPHONE_PORT", str(DEFAULT_PORT)))
CERT_PATH = args.cert or os.getenv("CLAWPHONE_CERT", DEFAULT_CERT_PATH)
KEY_PATH = args.key or os.getenv("CLAWPHONE_KEY", DEFAULT_KEY_PATH)
LOG_DIR = args.log_dir or os.getenv("CLAWPHONE_LOG_DIR", DEFAULT_LOG_PATH)
# Ensure log directory exists
os.makedirs(LOG_DIR, exist_ok=True)
# Ensure DB directory exists
db_dir = os.path.dirname(DB_PATH)
if db_dir:
os.makedirs(db_dir, exist_ok=True)
# Configure logging
def setup_logging(use_syslog: bool = False):
"""Configure logging with file rotation and optional syslog"""
# Logging
def setup_logging(use_syslog=False):
logger = logging.getLogger("clawphone")
logger.setLevel(logging.INFO)
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
console_format = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
console_handler.setFormatter(console_format)
logger.addHandler(console_handler)
if use_syslog:
try:
import handler as syslog_handler
syslog = logging.handlers.SysLogHandler(address='/dev/log')
syslog.setLevel(logging.INFO)
syslog_format = logging.Formatter("clawphone: %(message)s")
syslog.setFormatter(syslog_format)
logger.addHandler(syslog)
except Exception as e:
logger.warning(f"Could not setup syslog: {e}")
else:
# Rotating file handler (3 files)
file_handler = RotatingFileHandler(
f"{LOG_DIR}/clawphone.log",
maxBytes=10*1024*1024, # 10MB
backupCount=3
)
file_handler.setLevel(logging.INFO)
file_format = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
file_handler.setFormatter(file_format)
logger.addHandler(file_handler)
console = logging.StreamHandler(sys.stdout)
console.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(console)
if not use_syslog:
os.makedirs(LOG_DIR, exist_ok=True)
fileh = RotatingFileHandler(f"{LOG_DIR}/clawphone.log", maxBytes=10*1024*1024, backupCount=3)
fileh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(fileh)
return logger
logger = setup_logging(args.syslog)
def generate_token() -> str:
"""Generate a secure random token"""
return secrets.token_hex(32)
async def get_saved_token() -> Optional[str]:
"""Get token from database if saved"""
async with aiosqlite.connect(DB_PATH) as db:
cur = await db.execute("SELECT value FROM config WHERE key = 'server_token'")
row = await cur.fetchone()
return row[0] if row else None
async def save_token(token: str):
"""Save token to database"""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""")
await db.execute("INSERT OR REPLACE INTO config (key, value) VALUES ('server_token', ?)", (token,))
await db.commit()
async def initialize_token():
"""Initialize token based on priority: CLI > env > DB > generate"""
global API_TOKEN
# Priority 1: CLI argument
if args.token:
API_TOKEN = args.token
await save_token(API_TOKEN)
logger.info(f"Token set from CLI argument")
return
# Priority 2: Environment variable
env_token = os.getenv("CLAWPHONE_TOKEN")
if env_token:
API_TOKEN = env_token
await save_token(API_TOKEN)
logger.info(f"Token set from environment variable")
return
# Priority 3: Generate new token (with --generate-token)
if args.generate_token:
API_TOKEN = generate_token()
await save_token(API_TOKEN)
logger.info(f"Generated new token")
return
# Priority 4: Check database for saved token
saved_token = await get_saved_token()
if saved_token:
API_TOKEN = saved_token
logger.info(f"Token loaded from database")
return
# Priority 5: Generate token if nothing specified
API_TOKEN = generate_token()
await save_token(API_TOKEN)
logger.info(f"Auto-generated new token")
app = FastAPI(title="ClawPhone Job Queue")
HTTP_CLIENT = httpx.AsyncClient(verify=False, timeout=30.0) # HTTPS verify=False
RETRY_INTERVALS = [1, 2, 5, 10]
# In-memory agent registry (loaded from DB)
API_TOKEN = ""
HTTP_CLIENT = httpx.AsyncClient(verify=False, timeout=30.0)
AGENTS = {}
# Create FastMCP server
mcp = FastMCP("ClawPhone", host=HOST, port=PORT)
async def init_db():
"""Initialize database"""
# Ensure DB directory exists
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
async with aiosqlite.connect(DB_PATH) as db:
# Config table for server settings
await db.execute("""
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""")
# Agents table with large text fields for prompts
await db.execute("""
CREATE TABLE IF NOT EXISTS agents (
name TEXT PRIMARY KEY,
hook TEXT NOT NULL,
token TEXT,
capability_prompt TEXT,
skill_prompt TEXT,
registered_at TEXT NOT NULL
)
""")
# Jobs table - added reason field
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,
reason TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
retry_count INTEGER DEFAULT 0,
next_retry_at TEXT
)
""")
await db.execute("CREATE TABLE IF NOT EXISTS config (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
await db.execute("CREATE TABLE IF NOT EXISTS agents (name TEXT PRIMARY KEY, hook TEXT NOT NULL, token TEXT, capability_prompt TEXT, skill_prompt TEXT, registered_at TEXT NOT NULL)")
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, reason 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()
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, capability_prompt, skill_prompt FROM agents")
rows = await cur.fetchall()
AGENTS = {
r[0]: {
"hook": r[1],
"token": r[2],
"capability_prompt": r[3] or "",
"skill_prompt": r[4] or ""
}
for r in rows
}
logger.info(f"Loaded {len(AGENTS)} agents: {list(AGENTS.keys())}")
def verify_token(authorization: str = Header(None)) -> bool:
"""Verify Bearer token"""
if not API_TOKEN:
return True # No auth configured
AGENTS = {r[0]: {"hook": r[1], "token": r[2], "capability_prompt": r[3] or "", "skill_prompt": r[4] or ""} for r in rows}
logger.info(f"Loaded {len(AGENTS)} agents")
if not authorization:
raise HTTPException(status_code=401, detail="Missing Authorization header")
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid authorization format")
token = authorization[7:]
if token != API_TOKEN:
raise HTTPException(status_code=401, detail="Invalid token")
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():
logger.info(f"Using existing TLS certs")
return
# Ensure directory exists
os.makedirs(os.path.dirname(CERT_PATH), exist_ok=True)
logger.info(f"Generating self-signed TLS certificate")
key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "ZZ"),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Internet"),
x509.NameAttribute(NameOID.LOCALITY_NAME, "ClawPhone"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "OpenClaw"),
x509.NameAttribute(NameOID.COMMON_NAME, "ClawPhone MCP"),
])
now = datetime.now()
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))
logger.info(f"TLS certificate generated")
async def notify_hook(agent_name: str, event: str, data: dict) -> bool:
"""Send webhook to agent (HTTPS, verify=False)"""
async def notify_hook(agent_name, event, data):
if agent_name not in AGENTS:
logger.warning(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:
logger.warning(f"No hook URL for agent {agent_name}")
if not agent.get("hook"):
return False
headers = {}
if hook_token:
headers["Authorization"] = f"Bearer {hook_token}"
payload = {
"event": event,
"agent": agent_name,
"timestamp": datetime.now().isoformat(),
**data
}
headers = {"Authorization": f"Bearer {agent['token']}"} if agent.get("token") else {}
try:
r = await HTTP_CLIENT.post(hook_url, json=payload, headers=headers)
logger.info(f"Webhook to {agent_name} ({event}): {r.status_code}")
r = await HTTP_CLIENT.post(agent["hook"], json={"event": event, "agent": agent_name, "timestamp": datetime.now().isoformat(), **data}, headers=headers)
return r.status_code in (200, 201)
except Exception as e:
logger.error(f"Webhook failed: {e}")
except:
return False
async def check_pending_jobs():
"""Background task to retry pending jobs"""
while True:
try:
async def init_token():
global API_TOKEN
if args.token:
API_TOKEN = args.token
elif os.getenv("CLAWPHONE_TOKEN"):
API_TOKEN = os.getenv("CLAWPHONE_TOKEN")
else:
async with aiosqlite.connect(DB_PATH) as db:
now = datetime.now().isoformat()
cur = await db.execute(
"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 = 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(target, "job_failed", {"job_id": job_id, "sender": sender, "title": title})
cur = await db.execute("SELECT value FROM config WHERE key = 'server_token'")
row = await cur.fetchone()
if row:
API_TOKEN = row[0]
else:
# 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))
API_TOKEN = secrets.token_hex(32)
await db.execute("INSERT OR REPLACE INTO config VALUES ('server_token', ?)", (API_TOKEN,))
await db.commit()
except Exception as e:
logger.error(f"Background error: {e}")
await asyncio.sleep(30)
@asynccontextmanager
async def lifespan(app):
asyncio.create_task(check_pending_jobs())
yield
await HTTP_CLIENT.aclose()
app = FastAPI(lifespan=lifespan)
# ============================================================
# MCP TOOLS ENDPOINTS (JSON-RPC style)
# ============================================================
@app.get("/tools")
async def list_tools(authorization: str = Header(None)):
verify_token(authorization)
return {"tools": [
{"name": "register", "description": "Register this agent with the MCP server",
"inputSchema": {"type": "object", "properties": {
"hook": {}, "token": {}, "name": {}, "capability_prompt": {}, "skill_prompt": {}
}, "required": ["hook", "token"]}},
{"name": "list_hosts", "description": "List all registered agents", "inputSchema": {"type": "object", "properties": {}}},
{"name": "get_host_info", "description": "Get info about a specific agent",
"inputSchema": {"type": "object", "properties": {"name": {}}, "required": ["name"]}},
{"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 pending job",
"inputSchema": {"type": "object", "properties": {"job_id": {}, "agent": {}}, "required": ["job_id", "agent"]}},
{"name": "reject_job", "description": "Reject a pending job with a reason",
"inputSchema": {"type": "object", "properties": {"job_id": {}, "agent": {}, "reason": {}}, "required": ["job_id", "agent", "reason"]}},
{"name": "update_job_status", "description": "Update job status",
"inputSchema": {"type": "object", "properties": {"job_id": {}, "status": {}, "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"]}},
]}
logger.info(f"Token: {API_TOKEN[:16]}...")
@app.get("/tools/list_hosts")
async def list_hosts(authorization: str = Header(None)):
verify_token(authorization)
await load_agents()
hosts = {name: {"hook": info.get("hook"), "capability_prompt": info.get("capability_prompt", "")[:500], "registered": True} for name, info in AGENTS.items()}
return {"hosts": hosts}
@app.get("/tools/get_host_info")
async def get_host_info(name: str, authorization: str = Header(None)):
verify_token(authorization)
await load_agents()
if name not in AGENTS:
return {"error": f"Agent '{name}' not found"}
info = AGENTS[name]
return {"name": name, "hook": info.get("hook"), "capability_prompt": info.get("capability_prompt", ""), "skill_prompt": info.get("skill_prompt", ""), "registered": True}
@app.post("/tools/register")
async def tool_register(data: dict = Body(...), authorization: str = Header(None)):
verify_token(authorization)
hook = data.get("hook")
token = data.get("token", "")
capability_prompt = data.get("capability_prompt", "")
skill_prompt = data.get("skill_prompt", "")
if not hook:
return {"success": False, "error": "hook URL is required"}
name = data.get("name")
# MCP Tools
@mcp.tool()
async def register(hook: str, token: str, name: str = None) -> str:
"""Register an agent with a webhook URL"""
if not name:
import urllib.parse
try:
parsed = urllib.parse.urlparse(hook)
name = parsed.netloc.split('.')[0]
except:
name = "unknown"
if not hook.startswith("https://"):
hook = hook.replace("http://", "https://")
name = urllib.parse.urlparse(hook).netloc.split('.')[0] or "unknown"
async with aiosqlite.connect(DB_PATH) as db:
await db.execute("""INSERT OR REPLACE INTO agents (name, hook, token, capability_prompt, skill_prompt, registered_at) VALUES (?, ?, ?, ?, ?, ?)""",
(name, hook, token, capability_prompt, skill_prompt, datetime.now().isoformat()))
await db.execute("INSERT OR REPLACE INTO agents VALUES (?, ?, ?, ?, ?, ?)",
(name, hook, token, "", "", datetime.now().isoformat()))
await db.commit()
await load_agents()
logger.info(f"Registered agent: {name}")
return {"success": True, "agent": name, "hook": hook}
return json.dumps({"success": True, "agent": name})
@mcp.tool()
async def list_hosts() -> str:
"""List all registered agents"""
return json.dumps({"hosts": {n: {"hook": i.get("hook"), "registered": True} for n, i in AGENTS.items()}})
@app.post("/tools/post_job")
async def tool_post_job(data: dict, authorization: str = Header(None)):
verify_token(authorization)
sender = data.get("sender")
target_agent = data.get("target_agent")
title = data.get("title")
description = data.get("description", "")
@mcp.tool()
async def get_host_info(name: str) -> str:
"""Get information about a specific agent"""
if name not in AGENTS:
return json.dumps({"error": "Not found"})
return json.dumps({"name": name, "hook": AGENTS[name].get("hook"), "registered": True})
@mcp.tool()
async def post_job(sender: str, target_agent: str, title: str, description: str = "") -> str:
"""Post a new job to an agent"""
if not all([sender, target_agent, title]):
return {"success": False, "error": "sender, target_agent, and title are required"}
return json.dumps({"success": False, "error": "missing 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', NULL, NULL, ?, ?, 0, ?)",
(job_id, sender, target_agent, title, description, now, now, first_retry))
await db.execute("INSERT INTO jobs VALUES (?, ?, ?, ?, ?, 'pending', NULL, NULL, ?, ?, 0, NULL)",
(job_id, sender, target_agent, title, description, now, now))
await db.commit()
await notify_hook(target_agent, "new_job", {"job_id": job_id, "sender": sender, "title": title})
return json.dumps({"success": True, "job_id": job_id})
await notify_hook(target_agent, "new_job", {"job_id": job_id, "sender": sender, "title": title, "description": description})
logger.info(f"Posted job {job_id}: {title} -> {target_agent}")
return {"success": True, "job_id": job_id}
@app.post("/tools/claim_job")
async def tool_claim_job(data: dict, authorization: str = Header(None)):
verify_token(authorization)
job_id = data.get("job_id")
agent = data.get("agent")
@mcp.tool()
async def claim_job(job_id: str, agent: str) -> str:
"""Claim a job for an agent"""
if not job_id or not agent:
return {"success": False, "error": "job_id and agent are required"}
return json.dumps({"success": False, "error": "missing required"})
async with aiosqlite.connect(DB_PATH) as db:
cur = await db.execute("SELECT status, target_agent, sender FROM jobs WHERE id = ?", (job_id,))
cur = await db.execute("SELECT status, target_agent FROM jobs WHERE id = ?", (job_id,))
row = await cur.fetchone()
if not row:
return {"success": False, "error": "Job not found"}
return json.dumps({"success": False, "error": "Not found"})
if row[0] != "pending":
return {"success": False, "error": f"Job is {row[0]}"}
return json.dumps({"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))
return json.dumps({"success": False, "error": "Wrong agent"})
await db.execute("UPDATE jobs SET status='claimed', updated_at=? WHERE id=?", (datetime.now().isoformat(), job_id))
await db.commit()
return json.dumps({"success": True})
sender = row[2]
if sender:
await notify_hook(sender, "job_claimed", {"job_id": job_id, "agent": agent})
logger.info(f"Agent {agent} claimed job {job_id}")
return {"success": True}
@app.post("/tools/reject_job")
async def tool_reject_job(data: dict, authorization: str = Header(None)):
"""Reject a pending job with a reason"""
verify_token(authorization)
job_id = data.get("job_id")
agent = data.get("agent")
reason = data.get("reason", "")
@mcp.tool()
async def reject_job(job_id: str, agent: str, reason: str = "") -> str:
"""Reject a job"""
if not job_id or not agent:
return {"success": False, "error": "job_id and agent are required"}
if not reason:
return {"success": False, "error": "reason is required when rejecting a job"}
return json.dumps({"success": False, "error": "missing required"})
async with aiosqlite.connect(DB_PATH) as db:
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": "Job not found"}
if row[0] != "pending":
return {"success": False, "error": f"Job is {row[0]}, can only reject pending jobs"}
if row[1] != agent:
return {"success": False, "error": "Job is for a different agent"}
await db.execute("UPDATE jobs SET status='rejected', reason=?, updated_at=? WHERE id=?", (reason, datetime.now().isoformat(), job_id))
await db.commit()
return json.dumps({"success": True})
# Notify sender about rejection
sender = row[2]
if sender:
await notify_hook(sender, "job_rejected", {"job_id": job_id, "agent": agent, "reason": reason})
logger.info(f"Agent {agent} rejected job {job_id}: {reason}")
return {"success": True}
@app.post("/tools/update_job_status")
async def tool_update_job_status(data: dict, authorization: str = Header(None)):
verify_token(authorization)
job_id = data.get("job_id")
status = data.get("status")
agent = data.get("agent")
result = data.get("result", "")
@mcp.tool()
async def update_job_status(job_id: str, status: str, agent: str, result: str = "") -> str:
"""Update job status"""
if not all([job_id, status, agent]):
return {"success": False, "error": "job_id, status, and agent are required"}
return json.dumps({"success": False, "error": "missing required"})
async with aiosqlite.connect(DB_PATH) as db:
cur = await db.execute("SELECT status, target_agent, sender FROM jobs WHERE id = ?", (job_id,))
cur = await db.execute("SELECT target_agent FROM jobs WHERE id = ?", (job_id,))
row = await cur.fetchone()
if not row:
return {"success": False, "error": "Job not found"}
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))
return json.dumps({"success": False, "error": "Not found"})
if row[0] != agent:
return json.dumps({"success": False, "error": "Wrong agent"})
await db.execute("UPDATE jobs SET status=?, result=?, updated_at=? WHERE id=?",
(status, result, datetime.now().isoformat(), job_id))
await db.commit()
return json.dumps({"success": True})
sender = row[2]
if sender:
await notify_hook(sender, "job_status_changed", {"job_id": job_id, "status": status, "result": result})
logger.info(f"Job {job_id} status updated to {status} by {agent}")
return {"success": True}
@app.get("/tools/list_jobs")
async def tool_list_jobs(agent: str = None, status: str = None, authorization: str = Header(None)):
verify_token(authorization)
@mcp.tool()
async def list_jobs(agent: str = None, status: str = None) -> str:
"""List jobs, optionally filtered by agent and status"""
q, p = "SELECT * FROM jobs WHERE 1=1", []
if agent:
q += " AND (target_agent=? OR sender=?)"
......@@ -578,122 +215,39 @@ async def tool_list_jobs(agent: str = None, status: str = None, authorization: s
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]}
return json.dumps({"jobs": [dict(r) for r in rows]})
@app.get("/tools/get_job")
async def tool_get_job(job_id: str, authorization: str = Header(None)):
verify_token(authorization)
@mcp.tool()
async def get_job(job_id: str) -> str:
"""Get job details"""
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"}
# ============================================================
# HTTP WEBHOOK ENDPOINTS (mirror MCP tools)
# ============================================================
@app.post("/hook/register")
async def http_register(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
return await tool_register(data, authorization)
@app.post("/hook/list_hosts")
async def http_list_hosts(authorization: str = Header(None)):
verify_token(authorization)
return await list_hosts(authorization)
@app.post("/hook/get_host_info")
async def http_get_host_info(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
name = data.get("name")
return await get_host_info(name, authorization)
@app.post("/hook/post_job")
async def http_post_job(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
return await tool_post_job(data, authorization)
return json.dumps({"job": dict(row)}) if row else json.dumps({"error": "Not found"})
@app.post("/hook/claim_job")
async def http_claim_job(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
return await tool_claim_job(data, authorization)
@app.post("/hook/reject_job")
async def http_reject_job(request: Request, authorization: str = Header(None)):
"""HTTP webhook for reject_job"""
verify_token(authorization)
data = await request.json()
return await tool_reject_job(data, authorization)
@app.post("/hook/update_job_status")
async def http_update_job_status(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
return await tool_update_job_status(data, authorization)
@app.post("/hook/list_jobs")
async def http_list_jobs(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
agent = data.get("agent")
status = data.get("status")
return await tool_list_jobs(agent, status, authorization)
@app.post("/hook/get_job")
async def http_get_job(request: Request, authorization: str = Header(None)):
verify_token(authorization)
data = await request.json()
job_id = data.get("job_id")
return await tool_get_job(job_id, authorization)
@app.get("/health")
async def health():
return {"status": "ok", "agents": len(AGENTS), "token_configured": bool(API_TOKEN)}
if __name__ == "__main__":
# Ensure directories exist
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) if os.path.dirname(DB_PATH) else None
os.makedirs(os.path.dirname(CERT_PATH), exist_ok=True) if os.path.dirname(CERT_PATH) else None
os.makedirs(os.path.dirname(KEY_PATH), exist_ok=True) if os.path.dirname(KEY_PATH) else None
# Initialize DB first (creates tables)
asyncio.run(init_db())
# Initialize token after DB is ready
asyncio.run(initialize_token())
generate_self_signed_cert()
async def main():
await init_db()
await init_token()
logger.info(f"Starting ClawPhone MCP Server on {HOST}:{PORT}")
logger.info("="*60)
logger.info(f" SERVER TOKEN: {API_TOKEN}")
logger.info(" (Copy this token to connect agents)")
logger.info("="*60)
if args.syslog:
logger.info("Using syslog for logging")
if args.stdio:
logger.info("Running in stdio mode...")
await mcp.run_stdio_async()
elif args.http:
logger.info("Running in streamable HTTP mode...")
await mcp.run_streamable_http_async()
else:
logger.info(f"Logging to {LOG_DIR}/clawphone.log")
logger.info("Running in SSE mode...")
await mcp.run_sse_async(mount_path="/mcp")
uvicorn.run(app, host=HOST, port=PORT, ssl_keyfile=KEY_PATH, ssl_certfile=CERT_PATH)
if __name__ == "__main__":
asyncio.run(main())
\ No newline at end of file
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