""" Platform Bridge MCP Server =========================== Goose MCP extension that bridges local Goose with the cloud platform. Provides tools: - fetch_agent_config: Get agent configuration from platform - sync_knowledge_base: Sync knowledge base to local - report_usage: Report anonymized usage statistics - validate_license: Validate offline license - get_platform_status: Check platform health """ __version__ = "1.0.0" __name__ = "platform_bridge" import os import sys import json import time import hashlib import hmac import urllib.request import urllib.error from typing import Any, Dict, List, Optional class PlatformBridgeClient: """Client for the cloud platform API""" def __init__(self, api_url: str = None, user_id: str = None, api_key: str = None): self.api_url = api_url or os.environ.get( "PLATFORM_API_URL", "http://210.16.161.218:18080" ) self.user_id = user_id or os.environ.get("PLATFORM_USER_ID", "local-agent") self.api_key = api_key or os.environ.get("PLATFORM_API_KEY", "") self.base_path = "/app/api/admin/privacy" def _request(self, method: str, path: str, params: dict = None) -> dict: """Make an HTTP request to the platform API""" url = f"{self.api_url}{path}" if params: query_parts = [] for k, v in params.items(): if v is not None: query_parts.append(f"{k}={urllib.request.quote(str(v))}") if query_parts: url += "?" + "&".join(query_parts) req = urllib.request.Request(url, method=method) if self.api_key: req.add_header("Authorization", f"Bearer {self.api_key}") req.add_header("Accept", "application/json") try: with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode()) return data except urllib.error.HTTPError as e: return {"code": 1, "error": f"HTTP {e.code}: {e.reason}"} except Exception as e: return {"code": 1, "error": str(e)} def fetch_agent_config(self, agent_id: str = None) -> dict: """Fetch agent configuration from platform""" params = {"user_id": self.user_id} if agent_id: params["agent_id"] = agent_id return self._request("GET", f"{self.base_path}/local-agents", params) def sync_knowledge_base(self, kb_id: int, direction: str = "pull") -> dict: """Sync knowledge base between platform and local""" return self._request("POST", f"{self.base_path}/kb-sync/trigger", { "knowledge_base_id": kb_id, "sync_direction": direction, "user_id": self.user_id, }) def report_usage(self, metrics: dict, report_type: str = "daily") -> dict: """Submit anonymized usage report""" return self._request("POST", f"{self.base_path}/usage/report", { "metrics": json.dumps(metrics), "report_type": report_type, "user_id": self.user_id, }) def validate_license(self, license_key: str, device_fingerprint: str = None) -> dict: """Validate offline license""" params = {"license_key": license_key, "user_id": self.user_id} if device_fingerprint: params["device_fingerprint"] = device_fingerprint return self._request("POST", f"{self.base_path}/licenses/validate", params) def get_privacy_config(self) -> dict: """Get current privacy configuration""" return self._request("GET", f"{self.base_path}/config", {"user_id": self.user_id}) def get_bridge_config(self) -> dict: """Get Platform Bridge configuration for export""" return self._request("GET", f"{self.base_path}/bridge-config", {"user_id": self.user_id}) def refresh_license(self, license_id: int, extend_days: int = 7) -> dict: """Refresh an existing license""" return self._request("POST", f"{self.base_path}/licenses/{license_id}/refresh", { "extend_days": extend_days, }) def sync_all_knowledge_bases(self) -> List[dict]: """Sync all configured knowledge bases""" config = self.get_privacy_config() if config.get("code") != 0: return [{"error": "Failed to get config"}] data = config.get("data", {}) kb_list = data.get("sync_knowledge_bases", []) if isinstance(kb_list, str): try: kb_list = json.loads(kb_list) except json.JSONDecodeError: kb_list = [] results = [] for kb_id in (kb_list or []): result = self.sync_knowledge_base(kb_id) results.append({"kb_id": kb_id, "result": result}) return results def check_health(self) -> dict: """Check platform health and connectivity""" start = time.time() try: result = self._request("GET", f"{self.base_path}/stats") elapsed = time.time() - start return { "connected": True, "latency_ms": round(elapsed * 1000), "platform_version": "qihuaiflow-g5", "stats": result.get("data", {}), } except Exception as e: return { "connected": False, "error": str(e), } # ======================================================================== # MCP Server Interface (_stdio transport) # ======================================================================== class PlatformBridgeMCPServer: """MCP-compatible server using stdio transport""" def __init__(self): self.client = PlatformBridgeClient() self.tools = { "fetch_agent_config": { "name": "fetch_agent_config", "description": "Fetch agent configuration from the cloud platform", "parameters": { "agent_id": {"type": "string", "description": "Optional agent ID", "required": False} } }, "sync_knowledge_base": { "name": "sync_knowledge_base", "description": "Sync knowledge base documents from platform to local", "parameters": { "kb_id": {"type": "integer", "description": "Knowledge base ID", "required": True}, "direction": {"type": "string", "description": "Sync direction: pull/push/bidirectional", "required": False} } }, "report_usage": { "name": "report_usage", "description": "Report anonymized usage statistics to platform", "parameters": { "metrics": {"type": "object", "description": "Usage metrics JSON", "required": True}, "report_type": {"type": "string", "description": "Report type: daily/weekly/monthly/realtime", "required": False} } }, "validate_license": { "name": "validate_license", "description": "Validate offline license key", "parameters": { "license_key": {"type": "string", "description": "The license key to validate", "required": True}, "device_fingerprint": {"type": "string", "description": "Device fingerprint for binding", "required": False} } }, "get_platform_status": { "name": "get_platform_status", "description": "Check cloud platform connectivity and health", "parameters": {} }, "get_agent_list": { "name": "get_agent_list", "description": "List available agents from the platform", "parameters": {} }, "sync_all_kb": { "name": "sync_all_kb", "description": "Sync all configured knowledge bases", "parameters": {} }, "get_privacy_config": { "name": "get_privacy_config", "description": "Get current privacy mode configuration", "parameters": {} } } def _handle_request(self, request: dict) -> dict: """Process an MCP tool request""" tool_name = request.get("name", "") args = request.get("arguments", {}) or {} handler_map = { "fetch_agent_config": lambda: self.client.fetch_agent_config(args.get("agent_id")), "sync_knowledge_base": lambda: self.client.sync_knowledge_base( args["kb_id"], args.get("direction", "pull") ), "report_usage": lambda: self.client.report_usage( args["metrics"], args.get("report_type", "daily") ), "validate_license": lambda: self.client.validate_license( args["license_key"], args.get("device_fingerprint") ), "get_platform_status": lambda: self.client.check_health(), "get_agent_list": lambda: self.client.fetch_agent_config(), "sync_all_kb": lambda: self.client.sync_all_knowledge_bases(), "get_privacy_config": lambda: self.client.get_privacy_config(), } handler = handler_map.get(tool_name) if not handler: return {"error": f"Unknown tool: {tool_name}"} try: result = handler() return result if isinstance(result, dict) else {"result": result} except Exception as e: return {"error": str(e)} def handle_tools_list(self) -> dict: """Return list of available tools (MCP tools/list)""" return {"tools": list(self.tools.values())} def handle_tools_call(self, request: dict) -> dict: """Handle a tool call (MCP tools/call)""" tool_name = request.get("name", "") args = request.get("arguments", {}) or {} result = self._handle_request({"name": tool_name, "arguments": args}) return { "content": [ { "type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2) } ] } def run(self): """Run MCP server in stdio mode""" print("Platform Bridge MCP Server v1.0", file=sys.stderr) print(f"API URL: {self.client.api_url}", file=sys.stderr) print(f"User ID: {self.client.user_id}", file=sys.stderr) for line in sys.stdin: line = line.strip() if not line: continue try: request = json.loads(line) method = request.get("method", "") if method == "tools/list": response = self.handle_tools_list() elif method == "tools/call": response = self.handle_tools_call(request.get("params", {})) elif method == "initialize": response = { "protocolVersion": "2024-11-05", "serverInfo": {"name": "platform-bridge", "version": "1.0.0"}, "capabilities": {"tools": {}} } else: response = {"error": f"Unknown method: {method}"} if "id" in request: response["id"] = request["id"] response["jsonrpc"] = "2.0" print(json.dumps(response), flush=True) except json.JSONDecodeError as e: err_resp = {"jsonrpc": "2.0", "error": {"code": -32700, "message": f"Parse error: {e}"}} print(json.dumps(err_resp), flush=True) except Exception as e: err_resp = {"jsonrpc": "2.0", "error": {"code": -32603, "message": str(e)}} print(json.dumps(err_resp), flush=True) def main(): """Entry point for python -m platform_bridge.mcp_server""" server = PlatformBridgeMCPServer() server.run() if __name__ == "__main__": main()