diff --git a/backend/python/platform_bridge/__init__.py b/backend/python/platform_bridge/__init__.py
new file mode 100644
index 0000000..a0dd389
--- /dev/null
+++ b/backend/python/platform_bridge/__init__.py
@@ -0,0 +1,24 @@
+"""
+Platform Bridge - Cloud-to-Local Agent Bridge
+==============================================
+
+A Python package providing MCP server and sync client for bridging
+Goose (local agent) with the Qihu-AIFlow cloud platform.
+
+Modules:
+- mcp_server: MCP-compatible server for Goose (stdio transport)
+- sync_client: Periodic sync manager for KB, license, usage
+- __init__: Package interface
+"""
+
+from .mcp_server import PlatformBridgeClient, PlatformBridgeMCPServer, main
+from .sync_client import SyncManager, create_sync_manager
+
+__version__ = "1.0.0"
+__all__ = [
+    "PlatformBridgeClient",
+    "PlatformBridgeMCPServer",
+    "SyncManager",
+    "create_sync_manager",
+    "main",
+]
diff --git a/backend/python/platform_bridge/mcp_server.py b/backend/python/platform_bridge/mcp_server.py
new file mode 100644
index 0000000..3288c34
--- /dev/null
+++ b/backend/python/platform_bridge/mcp_server.py
@@ -0,0 +1,308 @@
+"""
+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()
diff --git a/backend/python/platform_bridge/sync_client.py b/backend/python/platform_bridge/sync_client.py
new file mode 100644
index 0000000..3902835
--- /dev/null
+++ b/backend/python/platform_bridge/sync_client.py
@@ -0,0 +1,133 @@
+"""
+Platform Bridge - Cloud-to-Local Agent Bridge
+==============================================
+WorkBuddy-compatible sync client for local Goose agent.
+
+Provides:
+- Periodic knowledge base sync
+- Usage metrics collection and reporting
+- License validation and refresh
+- Platform health monitoring
+"""
+import os
+import json
+import time
+import hashlib
+from datetime import datetime, timedelta
+
+# Import the MCP server client
+from .mcp_server import PlatformBridgeClient
+
+
+class SyncManager:
+    """Manages periodic synchronization between local agent and cloud platform"""
+
+    def __init__(self, client: PlatformBridgeClient = None):
+        self.client = client or PlatformBridgeClient()
+        self.sync_interval = 3600  # Default: 1 hour
+        self.last_sync = {}
+        self.last_license_check = None
+        self.last_usage_report = None
+        self.usage_buffer = []
+
+    def configure(self, config: dict):
+        """Configure sync parameters from privacy config"""
+        frequency_map = {
+            "realtime": 60,
+            "hourly": 3600,
+            "daily": 86400,
+            "manual": None,  # No auto sync
+        }
+        freq = config.get("sync_frequency", "daily")
+        self.sync_interval = frequency_map.get(freq, 86400)
+        self.client.user_id = config.get("user_id", self.client.user_id)
+
+    def should_sync_kb(self, kb_id: int) -> bool:
+        """Check if a KB needs syncing"""
+        last = self.last_sync.get(kb_id, 0)
+        return (time.time() - last) >= self.sync_interval
+
+    def sync_kb(self, kb_id: int) -> dict:
+        """Sync a single knowledge base"""
+        result = self.client.sync_knowledge_base(kb_id)
+        self.last_sync[kb_id] = time.time()
+        return result
+
+    def sync_all(self, kb_ids: list) -> list:
+        """Sync all configured knowledge bases"""
+        results = []
+        for kb_id in kb_ids:
+            if self.should_sync_kb(kb_id):
+                result = self.sync_kb(kb_id)
+                results.append({"kb_id": kb_id, "result": result})
+        return results
+
+    def check_license(self, license_key: str) -> dict:
+        """Check if license is still valid"""
+        now = time.time()
+        if self.last_license_check and (now - self.last_license_check) < 3600:
+            return {"valid": True, "cached": True}
+
+        result = self.client.validate_license(license_key)
+        self.last_license_check = now
+        return result
+
+    def track_usage(self, event_type: str, metadata: dict = None):
+        """Buffer a usage event for reporting"""
+        self.usage_buffer.append({
+            "type": event_type,
+            "timestamp": time.time(),
+            "metadata": metadata or {},
+        })
+
+    def report_usage(self) -> dict:
+        """Submit buffered usage events"""
+        if not self.usage_buffer:
+            return {"reported": 0}
+
+        events = self.usage_buffer[:]
+        self.usage_buffer = []
+
+        metrics = {
+            "event_count": len(events),
+            "event_types": {},
+            "time_range": {
+                "start": min(e["timestamp"] for e in events),
+                "end": max(e["timestamp"] for e in events),
+            }
+        }
+
+        for e in events:
+            t = e["type"]
+            metrics["event_types"][t] = metrics["event_types"].get(t, 0) + 1
+
+        result = self.client.report_usage(metrics)
+        self.last_usage_report = time.time()
+        result["reported"] = len(events)
+        return result
+
+    def health_check(self) -> dict:
+        """Run a full health check"""
+        platform = self.client.check_health()
+        config = self.client.get_privacy_config()
+
+        return {
+            "platform": platform,
+            "privacy_mode": config.get("data", {}).get("privacy_mode", "unknown"),
+            "pending_events": len(self.usage_buffer),
+            "last_license_check": self.last_license_check,
+            "last_usage_report": self.last_usage_report,
+        }
+
+
+def create_sync_manager(api_url: str = None, user_id: str = None) -> SyncManager:
+    """Factory function to create a configured SyncManager"""
+    client = PlatformBridgeClient(api_url=api_url, user_id=user_id)
+    manager = SyncManager(client)
+
+    # Load config
+    config_result = client.get_privacy_config()
+    if config_result.get("code") == 0:
+        manager.configure(config_result["data"])
+
+    return manager