diff --git a/backend/python/app/api/admin/mcp_admin.py b/backend/python/app/api/admin/mcp_admin.py
index 81e4457..4840480 100644
--- a/backend/python/app/api/admin/mcp_admin.py
+++ b/backend/python/app/api/admin/mcp_admin.py
@@ -1,449 +1,511 @@
-"""
-MCP (Model Context Protocol) Server Management API
-Phase 3.3: Manage MCP servers, discover tools/resources, test connections
-"""
-from fastapi import APIRouter, Depends, HTTPException, Query, Body
-from sqlalchemy import or_, func, desc, asc, text
-from sqlalchemy.orm import Session
-from pydantic import BaseModel, Field
-from typing import Optional, List, Dict, Any
-from datetime import datetime
-import uuid
-import json
-import logging
-
-from ...database import get_db
-from ...models import gen_uuid
-from ...auth import get_current_admin
-from ...middleware.log_service import LogService
-
-router = APIRouter()
-logger = logging.getLogger("aiflow")
-
-
-# ============================================================
-# Pydantic Models
-# ============================================================
-
-class MCPServerCreate(BaseModel):
-    name: str
-    description: Optional[str] = ""
-    transport_type: str = "stdio"
-    command: Optional[str] = ""
-    args: Optional[Any] = None
-    env_vars: Optional[Dict[str, str]] = None
-    url: Optional[str] = ""
-    headers: Optional[Dict[str, str]] = None
-    is_enabled: Optional[bool] = True
-
-
-class MCPServerUpdate(BaseModel):
-    name: Optional[str] = None
-    description: Optional[str] = None
-    transport_type: Optional[str] = None
-    command: Optional[str] = None
-    args: Optional[Any] = None
-    env_vars: Optional[Dict[str, str]] = None
-    url: Optional[str] = None
-    headers: Optional[Dict[str, str]] = None
-    is_enabled: Optional[bool] = None
-
-
-class MCPServerConnectionTest(BaseModel):
-    transport_type: Optional[str] = None
-    command: Optional[str] = None
-    args: Optional[Any] = None
-    env_vars: Optional[Dict[str, str]] = None
-    url: Optional[str] = None
-    headers: Optional[Dict[str, str]] = None
-
-
-# ============================================================
-# Helper Functions
-# ============================================================
-
-def _format_cst_time(dt):
-    """Format datetime to CST string"""
-    if dt is None:
-        return None
-    try:
-        from datetime import timezone, timedelta
-        CST = timezone(timedelta(hours=8))
-        if dt.tzinfo is None:
-            dt = dt.replace(tzinfo=CST)
-        return dt.astimezone(CST).strftime("%Y-%m-%d %H:%M:%S")
-    except Exception:
-        return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None
-
-
-def server_to_dict(server) -> dict:
-    """Convert MCPServer model to dict"""
-    return {
-        "id": server.id,
-        "name": server.name,
-        "description": server.description or "",
-        "transport_type": server.transport_type or "stdio",
-        "transport_type_label": {"stdio": "STDIO", "sse": "SSE"}.get(server.transport_type, server.transport_type),
-        "command": server.command or "",
-        "args": json.loads(server.args) if server.args else {},
-        "env_vars": json.loads(server.env_vars) if server.env_vars else {},
-        "url": server.url or "",
-        "headers": json.loads(server.headers) if server.headers else {},
-        "is_enabled": server.is_enabled if server.is_enabled is not None else True,
-        "status": server.status or "disconnected",
-        "last_connected_at": _format_cst_time(server.last_connected_at),
-        "tools_count": server.tools_count or 0,
-        "resources_count": server.resources_count or 0,
-        "created_at": _format_cst_time(server.created_at),
-        "updated_at": _format_cst_time(server.updated_at),
-    }
-
-
-# ============================================================
-# API Endpoints
-# ============================================================
-
-@router.get("/stats/overview", summary="Get MCP server statistics")
-async def get_mcp_stats(
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Get overview statistics for MCP servers"""
-    from ...models import MCPServer
-    total = db.query(func.count(MCPServer.id)).scalar() or 0
-    enabled = db.query(func.count(MCPServer.id)).filter(MCPServer.is_enabled == True).scalar() or 0
-    connected = db.query(func.count(MCPServer.id)).filter(MCPServer.status == "connected").scalar() or 0
-
-    # By transport type
-    by_transport = {}
-    rows = db.query(MCPServer.transport_type, func.count(MCPServer.id)).group_by(MCPServer.transport_type).all()
-    for t, c in rows:
-        by_transport[t or "unknown"] = c
-
-    # By status
-    by_status = {}
-    rows = db.query(MCPServer.status, func.count(MCPServer.id)).group_by(MCPServer.status).all()
-    for s, c in rows:
-        by_status[s or "unknown"] = c
-
-    return {
-        "code": 0,
-        "data": {
-            "total_servers": total,
-            "enabled_servers": enabled,
-            "connected_servers": connected,
-            "by_transport": by_transport,
-            "by_status": by_status,
-        },
-        "message": "ok",
-    }
-
-
-@router.get("/", summary="List MCP servers")
-async def list_mcp_servers(
-    page: int = Query(1, ge=1),
-    page_size: int = Query(20, ge=1, le=100),
-    search: Optional[str] = Query(None),
-    transport_type: Optional[str] = Query(None),
-    status: Optional[str] = Query(None),
-    is_enabled: Optional[bool] = Query(None),
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """List all MCP servers with pagination and filters"""
-    from ...models import MCPServer
-    query = db.query(MCPServer)
-
-    if search:
-        query = query.filter(
-            or_(MCPServer.name.contains(search), MCPServer.description.contains(search))
-        )
-    if transport_type:
-        query = query.filter(MCPServer.transport_type == transport_type)
-    if status:
-        query = query.filter(MCPServer.status == status)
-    if is_enabled is not None:
-        query = query.filter(MCPServer.is_enabled == is_enabled)
-
-    total = query.count()
-    servers = query.order_by(desc(MCPServer.created_at)).offset((page - 1) * page_size).limit(page_size).all()
-
-    return {
-        "code": 0,
-        "data": {
-            "total": total,
-            "page": page,
-            "page_size": page_size,
-            "items": [server_to_dict(s) for s in servers],
-        },
-        "message": "ok",
-    }
-
-
-@router.get("/{server_id}", summary="Get MCP server details")
-async def get_mcp_server(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Get a single MCP server by ID"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-    return {"code": 0, "data": server_to_dict(server), "message": "ok"}
-
-
-@router.post("/", summary="Create MCP server")
-async def create_mcp_server(
-    data: MCPServerCreate,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Register a new MCP server"""
-    from ...models import MCPServer
-
-    # Check name uniqueness
-    existing = db.query(MCPServer).filter(MCPServer.name == data.name).first()
-    if existing:
-        raise HTTPException(status_code=400, detail="MCP server with this name already exists")
-
-    now = datetime.now()
-    server = MCPServer(
-        id=gen_uuid(),
-        name=data.name,
-        description=data.description or "",
-        transport_type=data.transport_type,
-        command=data.command or "",
-        args=json.dumps(data.args) if data.args else "{}",
-        env_vars=json.dumps(data.env_vars) if data.env_vars else "{}",
-        url=data.url or "",
-        headers=json.dumps(data.headers) if data.headers else "{}",
-        is_enabled=data.is_enabled if data.is_enabled is not None else True,
-        status="disconnected",
-        created_at=now,
-        updated_at=now,
-    )
-    db.add(server)
-    db.commit()
-    db.refresh(server)
-
-    logger.info(f"MCP server created: {server.name} ({server.id})")
-    return {"code": 0, "data": server_to_dict(server), "message": "MCP server created"}
-
-
-@router.put("/{server_id}", summary="Update MCP server")
-async def update_mcp_server(
-    server_id: str,
-    data: MCPServerUpdate,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Update an existing MCP server"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    update_fields = data.model_dump(exclude_unset=True)
-
-    # Handle JSON fields
-    for json_field in ["args", "env_vars", "headers"]:
-        if json_field in update_fields and update_fields[json_field] is not None:
-            update_fields[json_field] = json.dumps(update_fields[json_field])
-
-    for key, value in update_fields.items():
-        if value is not None:
-            setattr(server, key, value)
-
-    server.updated_at = datetime.now()
-    db.commit()
-    db.refresh(server)
-
-    logger.info(f"MCP server updated: {server.name} ({server.id})")
-    return {"code": 0, "data": server_to_dict(server), "message": "MCP server updated"}
-
-
-@router.delete("/{server_id}", summary="Delete MCP server")
-async def delete_mcp_server(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Delete an MCP server"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    name = server.name
-    db.delete(server)
-    db.commit()
-
-    logger.info(f"MCP server deleted: {name} ({server_id})")
-    return {"code": 0, "data": None, "message": "MCP server deleted"}
-
-
-@router.put("/{server_id}/toggle", summary="Enable/Disable MCP server")
-async def toggle_mcp_server(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """Toggle MCP server enabled state"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    server.is_enabled = not server.is_enabled
-    server.updated_at = datetime.now()
-    db.commit()
-
-    logger.info(f"MCP server toggled: {server.name} -> enabled={server.is_enabled}")
-    return {
-        "code": 0,
-        "data": {"id": server.id, "is_enabled": server.is_enabled},
-        "message": "MCP server toggled",
-    }
-
-
-@router.post("/{server_id}/test", summary="Test MCP server connection")
-async def test_mcp_connection(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-    test_data: Optional[MCPServerConnectionTest] = None,
-):
-    """Test connection to an MCP server"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    transport = test_data.transport_type if test_data and test_data.transport_type else server.transport_type
-    result = {"success": False, "message": "", "details": {}}
-
-    try:
-        if transport == "sse":
-            sse_url = (test_data.url if test_data else None) or server.url
-            if not sse_url:
-                result["message"] = "SSE URL is required"
-                return {"code": 0, "data": result, "message": "ok"}
-
-            # Test SSE endpoint with a simple HTTP request
-            import httpx
-            try:
-                response = httpx.get(sse_url, timeout=10.0)
-                result["success"] = response.status_code < 500
-                result["message"] = f"SSE endpoint responded with {response.status_code}"
-                result["details"]["status_code"] = response.status_code
-                result["details"]["content_type"] = response.headers.get("content-type", "")
-            except httpx.ConnectError:
-                result["message"] = "Connection refused - server may be down"
-            except httpx.TimeoutException:
-                result["message"] = "Connection timed out"
-
-        elif transport == "stdio":
-            cmd = (test_data.command if test_data else None) or server.command
-            if not cmd:
-                result["message"] = "STDIO command is required"
-                return {"code": 0, "data": result, "message": "ok"}
-
-            # Basic command existence check (not actually executing MCP handshake)
-            import subprocess
-            try:
-                subprocess.run([cmd, "--version"], capture_output=True, timeout=10)
-                result["success"] = True
-                result["message"] = "Command found and executable"
-            except FileNotFoundError:
-                result["message"] = "Command not found in PATH"
-            except subprocess.TimeoutExpired:
-                result["message"] = "Command timed out"
-            except Exception as e:
-                result["message"] = f"Command test failed"
-                result["details"]["error"] = str(e)
-
-        # Update server status
-        server.status = "connected" if result["success"] else "error"
-        if result["success"]:
-            server.last_connected_at = datetime.now()
-        server.updated_at = datetime.now()
-        db.commit()
-
-    except Exception as e:
-        result["message"] = str(e)
-        result["details"]["error"] = str(e)
-        server.status = "error"
-        server.updated_at = datetime.now()
-        db.commit()
-
-    return {"code": 0, "data": result, "message": "ok"}
-
-
-@router.get("/{server_id}/tools", summary="List MCP server tools")
-async def list_mcp_tools(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """List tools exposed by an MCP server (placeholder/mock)"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    # In a real implementation, this would connect to the MCP server and call tools/list
-    # For now, return mock data based on common MCP tools
-    mock_tools = []
-    if server.transport_type == "stdio":
-        mock_tools = [
-            {"name": "read_file", "description": "Read contents of a file"},
-            {"name": "write_file", "description": "Write contents to a file"},
-            {"name": "execute_command", "description": "Execute a shell command"},
-            {"name": "search_content", "description": "Search for content in files"},
-        ]
-    elif server.transport_type == "sse":
-        mock_tools = [
-            {"name": "web_search", "description": "Search the web"},
-            {"name": "fetch_url", "description": "Fetch content from URL"},
-            {"name": "api_request", "description": "Make API requests"},
-        ]
-
-    return {
-        "code": 0,
-        "data": {
-            "server_name": server.name,
-            "tools": mock_tools,
-            "count": len(mock_tools),
-        },
-        "message": "ok",
-    }
-
-
-@router.get("/{server_id}/resources", summary="List MCP server resources")
-async def list_mcp_resources(
-    server_id: str,
-    db: Session = Depends(get_db),
-    admin_user=Depends(get_current_admin),
-):
-    """List resources exposed by an MCP server (placeholder/mock)"""
-    from ...models import MCPServer
-    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
-    if not server:
-        raise HTTPException(status_code=404, detail="MCP server not found")
-
-    mock_resources = [
-        {"uri": f"{server.name}://docs/readme", "name": "Documentation", "mime_type": "text/markdown"},
-        {"uri": f"{server.name}://config/settings", "name": "Configuration", "mime_type": "application/json"},
-    ]
-
-    return {
-        "code": 0,
-        "data": {
-            "server_name": server.name,
-            "resources": mock_resources,
-            "count": len(mock_resources),
-        },
-        "message": "ok",
-    }
+"""
+MCP (Model Context Protocol) Server Management API
+Phase 3.3: Manage MCP servers, discover tools/resources, test connections
+"""
+from fastapi import APIRouter, Depends, HTTPException, Query, Body
+from sqlalchemy import or_, func, desc, asc, text
+from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+from datetime import datetime
+import uuid
+import json
+import logging
+import re
+
+from ...database import get_db
+from ...models import gen_uuid
+from ...auth import get_current_admin
+from ...middleware.log_service import LogService
+
+router = APIRouter()
+logger = logging.getLogger("aiflow")
+
+
+# ============================================================
+# Pydantic Models
+# ============================================================
+
+class MCPServerCreate(BaseModel):
+    name: str
+    description: Optional[str] = ""
+    transport_type: str = "stdio"
+    command: Optional[str] = ""
+    args: Optional[Any] = None
+    env_vars: Optional[Dict[str, str]] = None
+    url: Optional[str] = ""
+    headers: Optional[Dict[str, str]] = None
+    is_enabled: Optional[bool] = True
+
+
+class MCPServerUpdate(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    transport_type: Optional[str] = None
+    command: Optional[str] = None
+    args: Optional[Any] = None
+    env_vars: Optional[Dict[str, str]] = None
+    url: Optional[str] = None
+    headers: Optional[Dict[str, str]] = None
+    is_enabled: Optional[bool] = None
+
+
+class MCPServerConnectionTest(BaseModel):
+    transport_type: Optional[str] = None
+    command: Optional[str] = None
+    args: Optional[Any] = None
+    env_vars: Optional[Dict[str, str]] = None
+    url: Optional[str] = None
+    headers: Optional[Dict[str, str]] = None
+
+
+# ============================================================
+# Helper Functions
+# ============================================================
+
+def _format_cst_time(dt):
+    """Format datetime to CST string"""
+    if dt is None:
+        return None
+    try:
+        from datetime import timezone, timedelta
+        CST = timezone(timedelta(hours=8))
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=CST)
+        return dt.astimezone(CST).strftime("%Y-%m-%d %H:%M:%S")
+    except Exception:
+        return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None
+
+
+def server_to_dict(server) -> dict:
+    """Convert MCPServer model to dict"""
+    return {
+        "id": server.id,
+        "name": server.name,
+        "description": server.description or "",
+        "transport_type": server.transport_type or "stdio",
+        "transport_type_label": {"stdio": "STDIO", "sse": "SSE"}.get(server.transport_type, server.transport_type),
+        "command": server.command or "",
+        "args": json.loads(server.args) if server.args else {},
+        "env_vars": json.loads(server.env_vars) if server.env_vars else {},
+        "url": server.url or "",
+        "headers": json.loads(server.headers) if server.headers else {},
+        "is_enabled": server.is_enabled if server.is_enabled is not None else True,
+        "status": server.status or "disconnected",
+        "last_connected_at": _format_cst_time(server.last_connected_at),
+        "tools_count": server.tools_count or 0,
+        "resources_count": server.resources_count or 0,
+        "created_at": _format_cst_time(server.created_at),
+        "updated_at": _format_cst_time(server.updated_at),
+    }
+
+
+def _extract_url_from_text(text: str) -> Optional[str]:
+    """Extract first http/https URL from text."""
+    if not text:
+        return None
+    match = re.search(r"https?://[^\s\"'<>\)\]\}]+", text)
+    return match.group(0) if match else None
+
+
+def _sync_market_mcp_to_servers(db: Session):
+    """
+    Synchronize MarketApp MCP records (type='mcp') into mcp_servers table.
+    Uses MarketApp.id as MCPServer.id so detail/edit endpoints can work directly.
+    """
+    from ...models import MarketApp, MCPServer
+
+    market_mcps = db.query(MarketApp).filter(
+        MarketApp.type == "mcp",
+        MarketApp.status == "published",
+    ).all()
+
+    existing_ids = {row[0] for row in db.query(MCPServer.id).all()}
+
+    for ma in market_mcps:
+        if ma.id in existing_ids:
+            continue
+
+        transport = ma.mcp_transport or "streamable_http"
+        code = ma.skill_content or ""
+        url = _extract_url_from_text(code) or _extract_url_from_text(ma.description or "") or ""
+        command = ""
+        if not url and ma.agent_endpoint:
+            url = ma.agent_endpoint
+
+        now = datetime.now()
+        server = MCPServer(
+            id=ma.id,
+            name=ma.name,
+            description=ma.description or "",
+            transport_type=transport,
+            command=command,
+            args="{}",
+            env_vars="{}",
+            url=url,
+            headers="{}",
+            is_enabled=True,
+            status="disconnected",
+            tools_count=0,
+            resources_count=0,
+            created_at=ma.created_at or now,
+            updated_at=ma.updated_at or now,
+        )
+        db.add(server)
+
+    if market_mcps:
+        db.commit()
+
+
+# ============================================================
+# API Endpoints
+# ============================================================
+
+@router.get("/stats/overview", summary="Get MCP server statistics")
+async def get_mcp_stats(
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Get overview statistics for MCP servers"""
+    from ...models import MCPServer
+    _sync_market_mcp_to_servers(db)
+    total = db.query(func.count(MCPServer.id)).scalar() or 0
+    enabled = db.query(func.count(MCPServer.id)).filter(MCPServer.is_enabled == True).scalar() or 0
+    connected = db.query(func.count(MCPServer.id)).filter(MCPServer.status == "connected").scalar() or 0
+
+    # By transport type
+    by_transport = {}
+    rows = db.query(MCPServer.transport_type, func.count(MCPServer.id)).group_by(MCPServer.transport_type).all()
+    for t, c in rows:
+        by_transport[t or "unknown"] = c
+
+    # By status
+    by_status = {}
+    rows = db.query(MCPServer.status, func.count(MCPServer.id)).group_by(MCPServer.status).all()
+    for s, c in rows:
+        by_status[s or "unknown"] = c
+
+    return {
+        "code": 0,
+        "data": {
+            "total_servers": total,
+            "enabled_servers": enabled,
+            "connected_servers": connected,
+            "by_transport": by_transport,
+            "by_status": by_status,
+        },
+        "message": "ok",
+    }
+
+
+@router.get("/", summary="List MCP servers")
+async def list_mcp_servers(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    search: Optional[str] = Query(None),
+    transport_type: Optional[str] = Query(None),
+    status: Optional[str] = Query(None),
+    is_enabled: Optional[bool] = Query(None),
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """List all MCP servers with pagination and filters"""
+    from ...models import MCPServer
+    _sync_market_mcp_to_servers(db)
+    query = db.query(MCPServer)
+
+    if search:
+        query = query.filter(
+            or_(MCPServer.name.contains(search), MCPServer.description.contains(search))
+        )
+    if transport_type:
+        query = query.filter(MCPServer.transport_type == transport_type)
+    if status:
+        query = query.filter(MCPServer.status == status)
+    if is_enabled is not None:
+        query = query.filter(MCPServer.is_enabled == is_enabled)
+
+    total = query.count()
+    servers = query.order_by(desc(MCPServer.created_at)).offset((page - 1) * page_size).limit(page_size).all()
+
+    return {
+        "code": 0,
+        "data": {
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "items": [server_to_dict(s) for s in servers],
+        },
+        "message": "ok",
+    }
+
+
+@router.get("/{server_id}", summary="Get MCP server details")
+async def get_mcp_server(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Get a single MCP server by ID"""
+    from ...models import MCPServer
+    _sync_market_mcp_to_servers(db)
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+    return {"code": 0, "data": server_to_dict(server), "message": "ok"}
+
+
+@router.post("/", summary="Create MCP server")
+async def create_mcp_server(
+    data: MCPServerCreate,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Register a new MCP server"""
+    from ...models import MCPServer
+
+    # Check name uniqueness
+    existing = db.query(MCPServer).filter(MCPServer.name == data.name).first()
+    if existing:
+        raise HTTPException(status_code=400, detail="MCP server with this name already exists")
+
+    now = datetime.now()
+    server = MCPServer(
+        id=gen_uuid(),
+        name=data.name,
+        description=data.description or "",
+        transport_type=data.transport_type,
+        command=data.command or "",
+        args=json.dumps(data.args) if data.args else "{}",
+        env_vars=json.dumps(data.env_vars) if data.env_vars else "{}",
+        url=data.url or "",
+        headers=json.dumps(data.headers) if data.headers else "{}",
+        is_enabled=data.is_enabled if data.is_enabled is not None else True,
+        status="disconnected",
+        created_at=now,
+        updated_at=now,
+    )
+    db.add(server)
+    db.commit()
+    db.refresh(server)
+
+    logger.info(f"MCP server created: {server.name} ({server.id})")
+    return {"code": 0, "data": server_to_dict(server), "message": "MCP server created"}
+
+
+@router.put("/{server_id}", summary="Update MCP server")
+async def update_mcp_server(
+    server_id: str,
+    data: MCPServerUpdate,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Update an existing MCP server"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    update_fields = data.model_dump(exclude_unset=True)
+
+    # Handle JSON fields
+    for json_field in ["args", "env_vars", "headers"]:
+        if json_field in update_fields and update_fields[json_field] is not None:
+            update_fields[json_field] = json.dumps(update_fields[json_field])
+
+    for key, value in update_fields.items():
+        if value is not None:
+            setattr(server, key, value)
+
+    server.updated_at = datetime.now()
+    db.commit()
+    db.refresh(server)
+
+    logger.info(f"MCP server updated: {server.name} ({server.id})")
+    return {"code": 0, "data": server_to_dict(server), "message": "MCP server updated"}
+
+
+@router.delete("/{server_id}", summary="Delete MCP server")
+async def delete_mcp_server(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Delete an MCP server"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    name = server.name
+    db.delete(server)
+    db.commit()
+
+    logger.info(f"MCP server deleted: {name} ({server_id})")
+    return {"code": 0, "data": None, "message": "MCP server deleted"}
+
+
+@router.put("/{server_id}/toggle", summary="Enable/Disable MCP server")
+async def toggle_mcp_server(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """Toggle MCP server enabled state"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    server.is_enabled = not server.is_enabled
+    server.updated_at = datetime.now()
+    db.commit()
+
+    logger.info(f"MCP server toggled: {server.name} -> enabled={server.is_enabled}")
+    return {
+        "code": 0,
+        "data": {"id": server.id, "is_enabled": server.is_enabled},
+        "message": "MCP server toggled",
+    }
+
+
+@router.post("/{server_id}/test", summary="Test MCP server connection")
+async def test_mcp_connection(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+    test_data: Optional[MCPServerConnectionTest] = None,
+):
+    """Test connection to an MCP server"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    transport = test_data.transport_type if test_data and test_data.transport_type else server.transport_type
+    result = {"success": False, "message": "", "details": {}}
+
+    try:
+        if transport == "sse":
+            sse_url = (test_data.url if test_data else None) or server.url
+            if not sse_url:
+                result["message"] = "SSE URL is required"
+                return {"code": 0, "data": result, "message": "ok"}
+
+            # Test SSE endpoint with a simple HTTP request
+            import httpx
+            try:
+                response = httpx.get(sse_url, timeout=10.0)
+                result["success"] = response.status_code < 500
+                result["message"] = f"SSE endpoint responded with {response.status_code}"
+                result["details"]["status_code"] = response.status_code
+                result["details"]["content_type"] = response.headers.get("content-type", "")
+            except httpx.ConnectError:
+                result["message"] = "Connection refused - server may be down"
+            except httpx.TimeoutException:
+                result["message"] = "Connection timed out"
+
+        elif transport == "stdio":
+            cmd = (test_data.command if test_data else None) or server.command
+            if not cmd:
+                result["message"] = "STDIO command is required"
+                return {"code": 0, "data": result, "message": "ok"}
+
+            # Basic command existence check (not actually executing MCP handshake)
+            import subprocess
+            try:
+                subprocess.run([cmd, "--version"], capture_output=True, timeout=10)
+                result["success"] = True
+                result["message"] = "Command found and executable"
+            except FileNotFoundError:
+                result["message"] = "Command not found in PATH"
+            except subprocess.TimeoutExpired:
+                result["message"] = "Command timed out"
+            except Exception as e:
+                result["message"] = f"Command test failed"
+                result["details"]["error"] = str(e)
+
+        # Update server status
+        server.status = "connected" if result["success"] else "error"
+        if result["success"]:
+            server.last_connected_at = datetime.now()
+        server.updated_at = datetime.now()
+        db.commit()
+
+    except Exception as e:
+        result["message"] = str(e)
+        result["details"]["error"] = str(e)
+        server.status = "error"
+        server.updated_at = datetime.now()
+        db.commit()
+
+    return {"code": 0, "data": result, "message": "ok"}
+
+
+@router.get("/{server_id}/tools", summary="List MCP server tools")
+async def list_mcp_tools(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """List tools exposed by an MCP server (placeholder/mock)"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    # In a real implementation, this would connect to the MCP server and call tools/list
+    # For now, return mock data based on common MCP tools
+    mock_tools = []
+    if server.transport_type == "stdio":
+        mock_tools = [
+            {"name": "read_file", "description": "Read contents of a file"},
+            {"name": "write_file", "description": "Write contents to a file"},
+            {"name": "execute_command", "description": "Execute a shell command"},
+            {"name": "search_content", "description": "Search for content in files"},
+        ]
+    elif server.transport_type == "sse":
+        mock_tools = [
+            {"name": "web_search", "description": "Search the web"},
+            {"name": "fetch_url", "description": "Fetch content from URL"},
+            {"name": "api_request", "description": "Make API requests"},
+        ]
+
+    return {
+        "code": 0,
+        "data": {
+            "server_name": server.name,
+            "tools": mock_tools,
+            "count": len(mock_tools),
+        },
+        "message": "ok",
+    }
+
+
+@router.get("/{server_id}/resources", summary="List MCP server resources")
+async def list_mcp_resources(
+    server_id: str,
+    db: Session = Depends(get_db),
+    admin_user=Depends(get_current_admin),
+):
+    """List resources exposed by an MCP server (placeholder/mock)"""
+    from ...models import MCPServer
+    server = db.query(MCPServer).filter(MCPServer.id == server_id).first()
+    if not server:
+        raise HTTPException(status_code=404, detail="MCP server not found")
+
+    mock_resources = [
+        {"uri": f"{server.name}://docs/readme", "name": "Documentation", "mime_type": "text/markdown"},
+        {"uri": f"{server.name}://config/settings", "name": "Configuration", "mime_type": "application/json"},
+    ]
+
+    return {
+        "code": 0,
+        "data": {
+            "server_name": server.name,
+            "resources": mock_resources,
+            "count": len(mock_resources),
+        },
+        "message": "ok",
+    }
+
diff --git a/backend/python/app/api/market.py b/backend/python/app/api/market.py
index 5d0bc62..bd48a57 100644
--- a/backend/python/app/api/market.py
+++ b/backend/python/app/api/market.py
@@ -1,771 +1,814 @@
-from fastapi import APIRouter, Depends, HTTPException, Query
-from sqlalchemy.orm import Session
-from sqlalchemy import and_, desc, asc, func
-from pydantic import BaseModel, Field
-from typing import Optional
-import uuid
-from datetime import datetime
-import httpx
-from urllib.parse import urlparse
-
-from ..database import get_db
-from ..models import (
-    User, MarketApp, AppPurchase, AppReview, AppInstallation, Order,
-    AdBanner, gen_uuid
-)
-from ..auth import get_current_user, get_current_user_id, get_current_user_id_optional
-from ..middleware.log_service import LogService
-
-router = APIRouter()
-
-
-# ===== Agent 接入前连通性 / 鉴权校验 =====
-def _normalize_dify_url(endpoint: str) -> str:
-    ep = endpoint.rstrip("/")
-    if ep.endswith("/chat-messages"):
-        return ep
-    if ep.endswith("/v1/chat-messages"):
-        return ep
-    if ep.endswith("/v1"):
-        return ep + "/chat-messages"
-    return ep + "/v1/chat-messages"
-
-
-def _short_err(e: Exception) -> str:
-    return str(e)[:160]
-
-
-def _post_follow(url: str, headers: dict, json_body: dict, timeout: float = 15.0):
-    """POST 并手动跟随 301/302/307/308 重定向(**保持 POST 方法与请求体**)。
-
-    背景：Dify 若填写 http:// 地址会 301 永久重定向到 https://；httpx 内建
-    follow_redirects 对 301/302 会把 POST 降级为 GET(浏览器行为)，导致对
-    /v1/chat-messages 变成 GET → 405 method_not_allowed。故手动跟随并保持 POST。
-    容器内 env 有坏代理(502)，必须 trust_env=False 直连。
-    """
-    cur = url
-    last = None
-    for _ in range(4):
-        last = httpx.post(cur, headers=headers, json=json_body, timeout=timeout,
-                          trust_env=False, follow_redirects=False)
-        if last.status_code in (301, 302, 307, 308):
-            loc = last.headers.get("location")
-            if not loc:
-                return last
-            cur = loc if loc.startswith("http") else str(httpx.URL(cur).join(loc))
-            continue
-        return last
-    return last
-
-
-def verify_agent_connectivity(platform: str, endpoint: str, api_key: str) -> dict:
-    """
-    接入 Agent 前的连通性 + 鉴权校验。
-    返回 {ok, verified, detail}
-      ok:       是否通过（dify=鉴权通过；n8n/coze/其他=地址可达）
-      verified: 是否做了真正的鉴权验证（dify 为 True，其余为 False）
-    """
-    ep = (endpoint or "").strip()
-    if not ep or not ep.startswith(("http://", "https://")):
-        return {"ok": False, "verified": False, "detail": "Endpoint 必须以 http:// 或 https:// 开头"}
-    platform = (platform or "dify").lower()
-
-    # 关键：容器内 env 有坏代理(502)，必须 trust_env=False 直连
-    if platform == "dify":
-        url = _normalize_dify_url(ep)
-        payload = {
-            "inputs": {},
-            "query": "连通性测试",
-            "user": "__verify__",
-            "response_mode": "blocking",
-        }
-        headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-        try:
-            r = _post_follow(url, headers, payload, timeout=15.0)
-        except Exception as e:
-            return {"ok": False, "verified": False, "detail": f"无法连接 Dify 地址：{_short_err(e)}"}
-        if r.status_code == 200:
-            return {"ok": True, "verified": True, "detail": "连通且 API Key 有效"}
-        if r.status_code in (401, 403):
-            return {"ok": False, "verified": False, "detail": "API Key 无效（401/403）"}
-        if r.status_code == 404:
-            return {"ok": False, "verified": False,
-                    "detail": "地址路径不正确（404），应为 .../v1/chat-messages"}
-        if r.status_code in (301, 302, 307, 308):
-            loc = r.headers.get("location") or ""
-            return {"ok": False, "verified": False,
-                    "detail": f"地址被重定向（{r.status_code}）到 {loc or '未知'}；"
-                              f"请直接填写重定向后的地址（通常把 http:// 改为 https://）"}
-        if r.status_code == 400:
-            # 400 需区分两种情况：
-            #  (a) 到达应用且鉴权通过，但 Dify 应用/工作流内部执行报错(invalid_param + Run failed)
-            #  (b) 纯参数错误
-            body = (r.text or "")
-            low = body.lower()
-            if ("run failed" in low or "error code" in low or "workflow" in low
-                    or "node" in low or "-500" in body):
-                return {"ok": True, "verified": True,
-                        "detail": "连通且 API Key 有效；但你的 Dify 应用内部执行报错："
-                                  f"{body[:180]}。这是你自建 Dify 应用/工作流本身的问题"
-                                  "（例如代码节点沙箱缺少 python3），请在 Dify 侧修复。"}
-            return {"ok": False, "verified": False, "detail": f"参数错误（400）：{body[:180]}"}
-        return {"ok": False, "verified": False, "detail": f"Dify 返回状态码 {r.status_code}"}
-
-    # n8n / coze / 其他：仅做可达性探测（HEAD，避免触发 webhook 副作用）
-    try:
-        r = httpx.head(ep, timeout=12.0, trust_env=False, follow_redirects=True)
-        if r.status_code < 500:
-            return {"ok": True, "verified": False,
-                    "detail": f"地址可连通（HTTP {r.status_code}），运行时再做协议级鉴权"}
-        return {"ok": False, "verified": False, "detail": f"服务返回 {r.status_code}，地址可能不可用"}
-    except Exception as e:
-        return {"ok": False, "verified": False, "detail": f"地址不可达：{_short_err(e)}"}
-
-
-class VerifyAgentRequest(BaseModel):
-    platform: Optional[str] = None
-    endpoint: str
-    api_key: str
-
-
-@router.post("/apps/verify-agent")
-def verify_agent(
-    req: VerifyAgentRequest,
-    current_user: User = Depends(get_current_user),
-):
-    """接入 Agent 前手动测试连通性 / 鉴权，供前台「测试连接」按钮调用。"""
-    result = verify_agent_connectivity(req.platform or "dify", req.endpoint, req.api_key)
-    return {"code": 0, "data": result, "message": "ok"}
-
-
-@router.get("/ads")
-def list_public_ads(db: Session = Depends(get_db)):
-    """
-    前台首页「精选推荐」广告位：返回已上架(active)广告，按排序升序，无需登录。
-    供 LandingPage 轮播调用，无鉴权依赖。
-    """
-    ads = (
-        db.query(AdBanner)
-        .filter(AdBanner.status == "active")
-        .order_by(AdBanner.sort_order.asc())
-        .all()
-    )
-    return [
-        {
-            "id": a.id,
-            "title": a.title,
-            "image_url": a.image_url,
-            "link_app_id": a.link_app_id,
-            "link_app_name": a.link_app_name,
-            "sort_order": a.sort_order,
-            "click_count": a.click_count,
-            "status": a.status,
-            "created_at": a.created_at.isoformat() if a.created_at else None,
-        }
-        for a in ads
-    ]
-
-
-class UploadAppRequest(BaseModel):
-    name: str = Field(min_length=1, max_length=200)
-    description: Optional[str] = None
-    icon: Optional[str] = None
-    type: Optional[str] = None  # skill, agent, mcp
-    source: Optional[str] = None
-    price: float = Field(default=0, ge=0)
-    download_url: Optional[str] = None
-    category: Optional[str] = None
-    tags: Optional[str] = None
-    is_member_free: Optional[bool] = False
-    join_promotion: Optional[bool] = False
-    agent_endpoint: Optional[str] = None
-    api_key: Optional[str] = None
-    platform: Optional[str] = None
-
-
-class PurchaseAppRequest(BaseModel):
-    amount: Optional[float] = None
-
-
-class ReviewAppRequest(BaseModel):
-    rating: int = Field(ge=1, le=5)
-    content: Optional[str] = None
-
-
-@router.get("/apps")
-def list_apps(
-    type: Optional[str] = Query(None),
-    source: Optional[str] = Query(None),
-    category: Optional[str] = Query(None),
-    search: Optional[str] = Query(None),
-    sort: str = Query(default="latest"),
-    page: int = Query(default=1, ge=1),
-    page_size: int = Query(default=20, ge=1, le=100),
-    db: Session = Depends(get_db),
-    user_id: Optional[str] = Depends(get_current_user_id_optional),
-):
-    query = db.query(MarketApp).filter(MarketApp.status == "published")
-
-    if type:
-        query = query.filter(MarketApp.type == type)
-    if source:
-        query = query.filter(MarketApp.source == source)
-    if category:
-        query = query.filter(MarketApp.type == category)
-    if search:
-        query = query.filter(
-            and_(
-                MarketApp.name.ilike(f"%{search}%")
-                | MarketApp.description.ilike(f"%{search}%")
-            )
-        )
-
-    if sort == "popular":
-        query = query.order_by(desc(MarketApp.install_count))
-    elif sort == "rating":
-        query = query.order_by(desc(MarketApp.rating_avg))
-    else:
-        query = query.order_by(desc(MarketApp.created_at))
-
-    total = query.count()
-    apps = query.offset((page - 1) * page_size).limit(page_size).all()
-
-    # 如果用户已登录，获取该用户已安装的应用ID集合（确保按用户隔离）
-    installed_app_ids = set()
-    if user_id:
-        installations = db.query(AppInstallation.app_id).filter(
-            AppInstallation.user_id == user_id
-        ).all()
-        installed_app_ids = {inst[0] for inst in installations}
-
-    # 查询已安装状态时始终带 user_id 过滤
-
-    result = []
-    for app in apps:
-        # 用户自己创建/导入的应用应始终视为已安装（避免安装记录缺失导致显示未安装）
-        is_user_owned = bool(user_id) and app.source == "user" and app.publisher_id == user_id
-        item = {
-            "id": app.id,
-            "name": app.name,
-            "description": app.description,
-            "icon": app.icon,
-            "type": app.type,
-            "source": app.source,
-            "publisher_id": app.publisher_id,
-            "publisher_name": app.publisher_name,
-            "is_member_free": app.is_member_free if app.is_member_free else False,
-            "join_promotion": app.join_promotion if app.join_promotion else False,
-            "status": app.status,
-            "price": float(app.price) if app.price else 0,
-            "install_count": app.install_count or 0,
-            "rating_avg": float(app.rating_avg) if app.rating_avg else 0,
-            "review_count": app.review_count or 0,
-            "created_at": str(app.created_at) if app.created_at else None,
-            "installed": (app.id in installed_app_ids or is_user_owned) if user_id else False,
-        }
-        result.append(item)
-
-    return {
-        "code": 0,
-        "data": {
-            "total": total,
-            "page": page,
-            "page_size": page_size,
-            "items": result,
-        },
-        "message": "ok",
-    }
-
-
-@router.get("/apps/{app_id}")
-def get_app(
-    app_id: str,
-    db: Session = Depends(get_db),
-    current_user: Optional[User] = Depends(get_current_user),
-):
-    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
-    if not app:
-        raise HTTPException(status_code=404, detail="应用不存在")
-
-    # 检查当前用户是否已购买（按用户隔离）
-    purchased = False
-    installed = False
-    if current_user:
-        purchase = db.query(AppPurchase).filter(
-            and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
-        ).first()
-        if purchase:
-            purchased = True
-            installed = purchase.status == "installed"
-        # 通过 AppInstallation 表检查是否已安装（按用户隔离）
-        installation = db.query(AppInstallation).filter(
-            and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
-        ).first()
-        if installation:
-            installed = True
-
-    return {
-        "code": 0,
-        "data": {
-            "id": app.id, "name": app.name, "description": app.description,
-            "icon": app.icon, "type": app.type, "source": app.source,
-            "type": app.type, 
-            "price": float(app.price) if app.price else 0,
-            "install_count": app.install_count or 0,
-            "rating_avg": float(app.rating_avg) if app.rating_avg else 0,
-            "review_count": app.review_count or 0,
-            "publisher_id": app.publisher_id,
-            "download_url": app.download_url,
-            "detail": app.detail,
-            "created_at": str(app.created_at),
-            "purchased": purchased,
-            "installed": installed,
-        },
-        "message": "ok",
-    }
-
-
-@router.post("/apps")
-def upload_app(
-    req: UploadAppRequest,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    # 接入 Agent 时先做连通性 / 鉴权校验，避免把无效连接存进库
-    if (req.type or "") == "agent" and req.agent_endpoint and req.api_key:
-        v = verify_agent_connectivity(req.platform or "dify", req.agent_endpoint, req.api_key)
-        if not v["ok"]:
-            raise HTTPException(
-                status_code=400,
-                detail=f"Agent 连接校验未通过：{v['detail']}",
-            )
-
-    # source 判定：管理员（admin 角色）通过前台市场导入/开发的 Agent/Skill/MCP
-    # 统一标记为官方（official），其余用户按请求或默认为 user。
-    if current_user.role == "admin":
-        app_source = "official"
-    elif req.source and req.source in ("official", "user"):
-        app_source = req.source
-    else:
-        app_source = "user"
-
-    app = MarketApp(
-        id=gen_uuid(),
-        name=req.name,
-        description=req.description,
-        icon=req.icon,
-        type=req.type if req.type in ("skill","agent","mcp") else "skill",
-        source=app_source,
-        platform=req.platform,
-        price=req.price,
-        install_count=0,
-        rating_avg=0,
-        review_count=0,
-        publisher_id=current_user.id,
-        publisher_name=current_user.username,
-        is_member_free=req.is_member_free if req.is_member_free is not None else False,
-        join_promotion=req.join_promotion if req.join_promotion is not None else False,
-        agent_endpoint=req.agent_endpoint,
-        agent_api_key=req.api_key,
-        status="published",  # 用户创建的应用直接发布
-        created_at=datetime.now(),
-        updated_at=datetime.now(),
-    )
-    db.add(app)
-    db.commit()
-    db.refresh(app)
-
-    # 用户创建/导入的应用自动标记为已安装
-    existing_install = db.query(AppInstallation).filter(
-        AppInstallation.user_id == current_user.id,
-        AppInstallation.app_id == app.id,
-    ).first()
-    if not existing_install:
-        installation = AppInstallation(
-            user_id=current_user.id,
-            app_id=app.id,
-            installed_at=datetime.now(),
-        )
-        db.add(installation)
-        db.commit()
-
-    LogService.log_operation(
-        user_id=current_user.id, username=current_user.username, action="upload_app",
-        target="market_app", target_id=app.id,
-        detail=f"上传应用 {app.name}",
-    )
-
-    return {
-        "code": 0,
-        "data": {
-            "id": app.id, "name": app.name, "status": app.status,
-            "created_at": str(app.created_at),
-        },
-        "message": "提交成功，等待审核",
-    }
-
-
-@router.post("/apps/{app_id}/purchase")
-def purchase_app(
-    app_id: str,
-    req: PurchaseAppRequest,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    app = db.query(MarketApp).filter(
-        and_(MarketApp.id == app_id, MarketApp.status == "published")
-    ).first()
-    if not app:
-        raise HTTPException(status_code=404, detail="应用不存在或未上架")
-
-    # 检查是否已购买
-    existing = db.query(AppPurchase).filter(
-        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
-    ).first()
-    if existing and existing.status in ("purchased", "installed"):
-        raise HTTPException(status_code=400, detail="您已购买该应用")
-
-    amount = req.amount or app.price
-
-    # 创建订单
-    order = Order(
-        id=gen_uuid(),
-        user_id=current_user.id,
-        order_type="app",
-        item_id=app_id,
-        item_name=app.name,
-        amount=amount,
-        status="pending",
-        created_at=datetime.now(),
-    )
-    db.add(order)
-    db.flush()
-
-    # 创建购买记录
-    purchase = AppPurchase(
-        id=gen_uuid(),
-        user_id=current_user.id,
-        app_id=app_id,
-        order_id=order.id,
-        status="purchased",
-        purchased_at=datetime.now(),
-    )
-    db.add(purchase)
-
-    # 更新应用下载量
-    app.install_count = (app.install_count or 0) + 1
-    app.updated_at = datetime.now()
-
-    db.commit()
-
-    return {
-        "code": 0,
-        "data": {
-            "order_id": order.id,
-            "purchase_id": purchase.id,
-            "amount": amount,
-            "status": "purchased",
-        },
-        "message": "购买成功",
-    }
-
-
-@router.post("/apps/{app_id}/install")
-def install_app(
-    app_id: str,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
-    if not app:
-        raise HTTPException(status_code=404, detail="应用不存在")
-
-    # 检查是否已安装（通过 AppInstallation 表）
-    existing_install = db.query(AppInstallation).filter(
-        and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
-    ).first()
-    if existing_install:
-        raise HTTPException(status_code=400, detail="应用已安装")
-
-    # 检查是否已购买（付费应用需要先购买）
-    purchase = db.query(AppPurchase).filter(
-        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
-    ).first()
-
-    # 对于付费应用，必须已购买才能安装
-    if float(app.price or 0) > 0:
-        if not purchase:
-            raise HTTPException(status_code=403, detail="请先购买应用")
-        if purchase.status == "installed":
-            raise HTTPException(status_code=400, detail="应用已安装")
-
-    # 创建安装记录
-    installation = AppInstallation(
-        id=gen_uuid(),
-        user_id=current_user.id,
-        app_id=app_id,
-        installed_at=datetime.now(),
-    )
-    db.add(installation)
-
-    # 如果有购买记录，更新状态为已安装
-    if purchase:
-        purchase.status = "installed"
-        purchase.installed_at = datetime.now()
-
-    # 更新应用安装量
-    app.install_count = (app.install_count or 0) + 1
-    app.updated_at = datetime.now()
-
-    db.commit()
-
-    return {
-        "code": 0,
-        "data": {"installation_id": installation.id, "status": "installed"},
-        "message": "安装成功",
-    }
-
-
-@router.delete("/apps/{app_id}/uninstall")
-def uninstall_app(
-    app_id: str,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    """卸载应用 - 删除当前用户的安装记录（按用户隔离）"""
-    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
-    if not app:
-        raise HTTPException(status_code=404, detail="应用不存在")
-
-    # 查找当前用户的安装记录（按用户隔离）
-    installation = db.query(AppInstallation).filter(
-        and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
-    ).first()
-    if not installation:
-        raise HTTPException(status_code=400, detail="应用未安装")
-
-    # 删除安装记录
-    db.delete(installation)
-
-    # 问题4修复: 将AppPurchase状态从installed重置为purchased
-    purchase = db.query(AppPurchase).filter(
-        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
-    ).first()
-    if purchase and purchase.status == "installed":
-        purchase.status = "purchased"
-
-    # 更新应用安装量（减1，最小为0）
-    if app.install_count and app.install_count > 0:
-        app.install_count = app.install_count - 1
-    app.updated_at = datetime.now()
-
-    db.commit()
-
-    LogService.log_operation(
-        user_id=current_user.id, username=current_user.username, action="uninstall_app",
-        target="market_app", target_id=app_id,
-        detail=f"卸载应用 {app.name}",
-    )
-
-    return {
-        "code": 0,
-        "message": "卸载成功",
-    }
-
-
-@router.post("/apps/{app_id}/review")
-def review_app(
-    app_id: str,
-    req: ReviewAppRequest,
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
-    if not app:
-        raise HTTPException(status_code=404, detail="应用不存在")
-
-    # 检查是否已评价
-    existing = db.query(AppReview).filter(
-        and_(AppReview.app_id == app_id, AppReview.user_id == current_user.id)
-    ).first()
-    if existing:
-        # 更新已有评价
-        old_rating = existing.rating
-        existing.rating = req.rating
-        existing.content = req.content
-        existing.updated_at = datetime.now()
-
-        # 重新计算平均分 (总数不变，减去旧评分加新评分)
-        if app.review_count and app.review_count > 0:
-            total_rating = float(app.rating_avg or 0) * app.review_count
-            total_rating = total_rating - old_rating + req.rating
-            app.rating_avg = total_rating / app.review_count
-        else:
-            app.rating_avg = req.rating
-            app.review_count = 1
-    else:
-        # 创建新评价
-        review = AppReview(
-            id=gen_uuid(),
-            app_id=app_id,
-            user_id=current_user.id,
-            rating=req.rating,
-            content=req.content,
-            created_at=datetime.now(),
-            updated_at=datetime.now(),
-        )
-        db.add(review)
-
-        # 重新计算平均分
-        app.review_count = (app.review_count or 0) + 1
-        if app.review_count == 1:
-            app.rating_avg = req.rating
-        else:
-            total_rating = float(app.rating_avg or 0) * (app.review_count - 1) + req.rating
-            app.rating_avg = round(total_rating / app.review_count, 1)
-
-    app.updated_at = datetime.now()
-    db.commit()
-
-    return {
-        "code": 0,
-        "data": {
-            "rating_avg": float(app.rating_avg or 0),
-            "review_count": app.review_count or 0,
-        },
-        "message": "评价成功",
-    }
-
-
-@router.get("/my-apps")
-def get_my_apps(
-    status: Optional[str] = Query(None),
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    query = db.query(AppPurchase).filter(AppPurchase.user_id == current_user.id)
-    purchases = query.order_by(desc(AppPurchase.created_at)).all()
-
-    result = []
-    for p in purchases:
-        app = db.query(MarketApp).filter(MarketApp.id == p.app_id).first()
-        if app:
-            result.append({
-                "id": p.id,
-                "app_id": app.id,
-                "name": app.name,
-                "icon": app.icon,
-                "type": app.type,
-                "price": float(app.price or 0),
-                "purchased_at": str(p.created_at) if p.created_at else None,
-            })
-
-    return {"code": 0, "data": result, "message": "ok"}
-
-
-@router.get("/reviews/{app_id}")
-def get_reviews(
-    app_id: str,
-    page: int = Query(default=1, ge=1),
-    page_size: int = Query(default=20, ge=1, le=100),
-    db: Session = Depends(get_db),
-):
-    query = db.query(AppReview).filter(AppReview.app_id == app_id)
-    total = query.count()
-
-    reviews = (
-        query.order_by(desc(AppReview.created_at))
-        .offset((page - 1) * page_size)
-        .limit(page_size)
-        .all()
-    )
-
-    result = []
-    for r in reviews:
-        user = db.query(User).filter(User.id == r.user_id).first()
-        result.append({
-            "id": r.id,
-            "user_id": r.user_id,
-            "username": user.username if user else "unknown",
-            "avatar": user.avatar if user else None,
-            "rating": r.rating,
-            "content": r.content,
-            "created_at": str(r.created_at) if r.created_at else None,
-        })
-
-    return {
-        "code": 0,
-        "data": {"total": total, "page": page, "page_size": page_size, "items": result},
-        "message": "ok",
-    }
-
-
-# ============================================================
-# Agent部署资源查询接口
-# ============================================================
-
-@router.get("/agent-resources")
-def get_agent_resources(
-    db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
-):
-    """Agent部署时需要的资源列表：可用的应用、知识库、文档"""
-    # 用户已发布的应用
-    my_apps = db.query(MarketApp).filter(
-        MarketApp.publisher_id == current_user.id,
-        MarketApp.status == "published",
-    ).order_by(desc(MarketApp.updated_at)).all()
-
-    apps_result = []
-    for app in my_apps:
-        apps_result.append({
-            "id": app.id,
-            "name": app.name,
-            "type": app.type,
-            "description": app.description,
-            "install_count": app.install_count or 0,
-        })
-
-    # 用户知识库
-    from ..models import KnowledgeBase, KnowledgeDocument
-    kb_list = db.query(KnowledgeBase).filter(
-        KnowledgeBase.user_id == current_user.id
-    ).order_by(desc(KnowledgeBase.updated_at)).all()
-
-    kb_result = []
-    for kb in kb_list:
-        doc_count = db.query(KnowledgeDocument).filter(
-            KnowledgeDocument.knowledge_base_id == kb.id
-        ).count()
-        kb_result.append({
-            "id": kb.id,
-            "name": kb.name,
-            "description": kb.description,
-            "document_count": doc_count,
-        })
-
-    # 用户已安装的应用（从AppPurchase）
-    installed = db.query(AppPurchase).filter(
-        AppPurchase.user_id == current_user.id
-    ).all()
-    installed_ids = [p.app_id for p in installed]
-    installed_apps = []
-    if installed_ids:
-        iapps = db.query(MarketApp).filter(MarketApp.id.in_(installed_ids)).all()
-        for ia in iapps:
-            installed_apps.append({
-                "id": ia.id,
-                "name": ia.name,
-                "type": ia.type,
-                "description": ia.description,
-            })
-
-    return {
-        "code": 0,
-        "data": {
-            "my_apps": apps_result,
-            "knowledge_bases": kb_result,
-            "installed_apps": installed_apps,
-        },
-        "message": "ok",
-    }
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from sqlalchemy import and_, desc, asc, func
+from pydantic import BaseModel, Field
+from typing import Optional
+import uuid
+from datetime import datetime
+import httpx
+from urllib.parse import urlparse
+import re
+
+from ..database import get_db
+from ..models import (
+    User, MarketApp, AppPurchase, AppReview, AppInstallation, Order,
+    AdBanner, MCPServer, gen_uuid
+)
+from ..auth import get_current_user, get_current_user_id, get_current_user_id_optional
+from ..middleware.log_service import LogService
+
+router = APIRouter()
+
+
+def _extract_url_from_text(text: Optional[str]) -> Optional[str]:
+    """Extract first http/https URL from text."""
+    if not text:
+        return None
+    match = re.search(r"https?://[^\s\"'<>\)\]\}]+", text)
+    return match.group(0) if match else None
+
+
+# ===== Agent 接入前连通性 / 鉴权校验 =====
+def _normalize_dify_url(endpoint: str) -> str:
+    ep = endpoint.rstrip("/")
+    if ep.endswith("/chat-messages"):
+        return ep
+    if ep.endswith("/v1/chat-messages"):
+        return ep
+    if ep.endswith("/v1"):
+        return ep + "/chat-messages"
+    return ep + "/v1/chat-messages"
+
+
+def _short_err(e: Exception) -> str:
+    return str(e)[:160]
+
+
+def _post_follow(url: str, headers: dict, json_body: dict, timeout: float = 15.0):
+    """POST 并手动跟随 301/302/307/308 重定向(**保持 POST 方法与请求体**)。
+
+    背景：Dify 若填写 http:// 地址会 301 永久重定向到 https://；httpx 内建
+    follow_redirects 对 301/302 会把 POST 降级为 GET(浏览器行为)，导致对
+    /v1/chat-messages 变成 GET → 405 method_not_allowed。故手动跟随并保持 POST。
+    容器内 env 有坏代理(502)，必须 trust_env=False 直连。
+    """
+    cur = url
+    last = None
+    for _ in range(4):
+        last = httpx.post(cur, headers=headers, json=json_body, timeout=timeout,
+                          trust_env=False, follow_redirects=False)
+        if last.status_code in (301, 302, 307, 308):
+            loc = last.headers.get("location")
+            if not loc:
+                return last
+            cur = loc if loc.startswith("http") else str(httpx.URL(cur).join(loc))
+            continue
+        return last
+    return last
+
+
+def verify_agent_connectivity(platform: str, endpoint: str, api_key: str) -> dict:
+    """
+    接入 Agent 前的连通性 + 鉴权校验。
+    返回 {ok, verified, detail}
+      ok:       是否通过（dify=鉴权通过；n8n/coze/其他=地址可达）
+      verified: 是否做了真正的鉴权验证（dify 为 True，其余为 False）
+    """
+    ep = (endpoint or "").strip()
+    if not ep or not ep.startswith(("http://", "https://")):
+        return {"ok": False, "verified": False, "detail": "Endpoint 必须以 http:// 或 https:// 开头"}
+    platform = (platform or "dify").lower()
+
+    # 关键：容器内 env 有坏代理(502)，必须 trust_env=False 直连
+    if platform == "dify":
+        url = _normalize_dify_url(ep)
+        payload = {
+            "inputs": {},
+            "query": "连通性测试",
+            "user": "__verify__",
+            "response_mode": "blocking",
+        }
+        headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+        try:
+            r = _post_follow(url, headers, payload, timeout=15.0)
+        except Exception as e:
+            return {"ok": False, "verified": False, "detail": f"无法连接 Dify 地址：{_short_err(e)}"}
+        if r.status_code == 200:
+            return {"ok": True, "verified": True, "detail": "连通且 API Key 有效"}
+        if r.status_code in (401, 403):
+            return {"ok": False, "verified": False, "detail": "API Key 无效（401/403）"}
+        if r.status_code == 404:
+            return {"ok": False, "verified": False,
+                    "detail": "地址路径不正确（404），应为 .../v1/chat-messages"}
+        if r.status_code in (301, 302, 307, 308):
+            loc = r.headers.get("location") or ""
+            return {"ok": False, "verified": False,
+                    "detail": f"地址被重定向（{r.status_code}）到 {loc or '未知'}；"
+                              f"请直接填写重定向后的地址（通常把 http:// 改为 https://）"}
+        if r.status_code == 400:
+            # 400 需区分两种情况：
+            #  (a) 到达应用且鉴权通过，但 Dify 应用/工作流内部执行报错(invalid_param + Run failed)
+            #  (b) 纯参数错误
+            body = (r.text or "")
+            low = body.lower()
+            if ("run failed" in low or "error code" in low or "workflow" in low
+                    or "node" in low or "-500" in body):
+                return {"ok": True, "verified": True,
+                        "detail": "连通且 API Key 有效；但你的 Dify 应用内部执行报错："
+                                  f"{body[:180]}。这是你自建 Dify 应用/工作流本身的问题"
+                                  "（例如代码节点沙箱缺少 python3），请在 Dify 侧修复。"}
+            return {"ok": False, "verified": False, "detail": f"参数错误（400）：{body[:180]}"}
+        return {"ok": False, "verified": False, "detail": f"Dify 返回状态码 {r.status_code}"}
+
+    # n8n / coze / 其他：仅做可达性探测（HEAD，避免触发 webhook 副作用）
+    try:
+        r = httpx.head(ep, timeout=12.0, trust_env=False, follow_redirects=True)
+        if r.status_code < 500:
+            return {"ok": True, "verified": False,
+                    "detail": f"地址可连通（HTTP {r.status_code}），运行时再做协议级鉴权"}
+        return {"ok": False, "verified": False, "detail": f"服务返回 {r.status_code}，地址可能不可用"}
+    except Exception as e:
+        return {"ok": False, "verified": False, "detail": f"地址不可达：{_short_err(e)}"}
+
+
+class VerifyAgentRequest(BaseModel):
+    platform: Optional[str] = None
+    endpoint: str
+    api_key: str
+
+
+@router.post("/apps/verify-agent")
+def verify_agent(
+    req: VerifyAgentRequest,
+    current_user: User = Depends(get_current_user),
+):
+    """接入 Agent 前手动测试连通性 / 鉴权，供前台「测试连接」按钮调用。"""
+    result = verify_agent_connectivity(req.platform or "dify", req.endpoint, req.api_key)
+    return {"code": 0, "data": result, "message": "ok"}
+
+
+@router.get("/ads")
+def list_public_ads(db: Session = Depends(get_db)):
+    """
+    前台首页「精选推荐」广告位：返回已上架(active)广告，按排序升序，无需登录。
+    供 LandingPage 轮播调用，无鉴权依赖。
+    """
+    ads = (
+        db.query(AdBanner)
+        .filter(AdBanner.status == "active")
+        .order_by(AdBanner.sort_order.asc())
+        .all()
+    )
+    return [
+        {
+            "id": a.id,
+            "title": a.title,
+            "image_url": a.image_url,
+            "link_app_id": a.link_app_id,
+            "link_app_name": a.link_app_name,
+            "sort_order": a.sort_order,
+            "click_count": a.click_count,
+            "status": a.status,
+            "created_at": a.created_at.isoformat() if a.created_at else None,
+        }
+        for a in ads
+    ]
+
+
+class UploadAppRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=200)
+    description: Optional[str] = None
+    icon: Optional[str] = None
+    type: Optional[str] = None  # skill, agent, mcp
+    source: Optional[str] = None
+    price: float = Field(default=0, ge=0)
+    download_url: Optional[str] = None
+    category: Optional[str] = None
+    tags: Optional[str] = None
+    is_member_free: Optional[bool] = False
+    join_promotion: Optional[bool] = False
+    agent_endpoint: Optional[str] = None
+    api_key: Optional[str] = None
+    platform: Optional[str] = None
+
+
+class PurchaseAppRequest(BaseModel):
+    amount: Optional[float] = None
+
+
+class ReviewAppRequest(BaseModel):
+    rating: int = Field(ge=1, le=5)
+    content: Optional[str] = None
+
+
+@router.get("/apps")
+def list_apps(
+    type: Optional[str] = Query(None),
+    source: Optional[str] = Query(None),
+    category: Optional[str] = Query(None),
+    search: Optional[str] = Query(None),
+    sort: str = Query(default="latest"),
+    page: int = Query(default=1, ge=1),
+    page_size: int = Query(default=20, ge=1, le=100),
+    db: Session = Depends(get_db),
+    user_id: Optional[str] = Depends(get_current_user_id_optional),
+):
+    query = db.query(MarketApp).filter(MarketApp.status == "published")
+
+    if type:
+        query = query.filter(MarketApp.type == type)
+    if source:
+        query = query.filter(MarketApp.source == source)
+    if category:
+        query = query.filter(MarketApp.type == category)
+    if search:
+        query = query.filter(
+            and_(
+                MarketApp.name.ilike(f"%{search}%")
+                | MarketApp.description.ilike(f"%{search}%")
+            )
+        )
+
+    if sort == "popular":
+        query = query.order_by(desc(MarketApp.install_count))
+    elif sort == "rating":
+        query = query.order_by(desc(MarketApp.rating_avg))
+    else:
+        query = query.order_by(desc(MarketApp.created_at))
+
+    total = query.count()
+    apps = query.offset((page - 1) * page_size).limit(page_size).all()
+
+    # 如果用户已登录，获取该用户已安装的应用ID集合（确保按用户隔离）
+    installed_app_ids = set()
+    if user_id:
+        installations = db.query(AppInstallation.app_id).filter(
+            AppInstallation.user_id == user_id
+        ).all()
+        installed_app_ids = {inst[0] for inst in installations}
+
+    # 查询已安装状态时始终带 user_id 过滤
+
+    result = []
+    for app in apps:
+        # 用户自己创建/导入的应用应始终视为已安装（避免安装记录缺失导致显示未安装）
+        is_user_owned = bool(user_id) and app.source == "user" and app.publisher_id == user_id
+        item = {
+            "id": app.id,
+            "name": app.name,
+            "description": app.description,
+            "icon": app.icon,
+            "type": app.type,
+            "source": app.source,
+            "publisher_id": app.publisher_id,
+            "publisher_name": app.publisher_name,
+            "is_member_free": app.is_member_free if app.is_member_free else False,
+            "join_promotion": app.join_promotion if app.join_promotion else False,
+            "status": app.status,
+            "price": float(app.price) if app.price else 0,
+            "install_count": app.install_count or 0,
+            "rating_avg": float(app.rating_avg) if app.rating_avg else 0,
+            "review_count": app.review_count or 0,
+            "created_at": str(app.created_at) if app.created_at else None,
+            "installed": (app.id in installed_app_ids or is_user_owned) if user_id else False,
+        }
+        result.append(item)
+
+    return {
+        "code": 0,
+        "data": {
+            "total": total,
+            "page": page,
+            "page_size": page_size,
+            "items": result,
+        },
+        "message": "ok",
+    }
+
+
+@router.get("/apps/{app_id}")
+def get_app(
+    app_id: str,
+    db: Session = Depends(get_db),
+    current_user: Optional[User] = Depends(get_current_user),
+):
+    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
+    if not app:
+        raise HTTPException(status_code=404, detail="应用不存在")
+
+    # 检查当前用户是否已购买（按用户隔离）
+    purchased = False
+    installed = False
+    if current_user:
+        purchase = db.query(AppPurchase).filter(
+            and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
+        ).first()
+        if purchase:
+            purchased = True
+            installed = purchase.status == "installed"
+        # 通过 AppInstallation 表检查是否已安装（按用户隔离）
+        installation = db.query(AppInstallation).filter(
+            and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
+        ).first()
+        if installation:
+            installed = True
+
+    return {
+        "code": 0,
+        "data": {
+            "id": app.id, "name": app.name, "description": app.description,
+            "icon": app.icon, "type": app.type, "source": app.source,
+            "type": app.type, 
+            "price": float(app.price) if app.price else 0,
+            "install_count": app.install_count or 0,
+            "rating_avg": float(app.rating_avg) if app.rating_avg else 0,
+            "review_count": app.review_count or 0,
+            "publisher_id": app.publisher_id,
+            "download_url": app.download_url,
+            "detail": app.detail,
+            "created_at": str(app.created_at),
+            "purchased": purchased,
+            "installed": installed,
+        },
+        "message": "ok",
+    }
+
+
+@router.post("/apps")
+def upload_app(
+    req: UploadAppRequest,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    # 接入 Agent 时先做连通性 / 鉴权校验，避免把无效连接存进库
+    if (req.type or "") == "agent" and req.agent_endpoint and req.api_key:
+        v = verify_agent_connectivity(req.platform or "dify", req.agent_endpoint, req.api_key)
+        if not v["ok"]:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Agent 连接校验未通过：{v['detail']}",
+            )
+
+    # source 判定：管理员（admin 角色）通过前台市场导入/开发的 Agent/Skill/MCP
+    # 统一标记为官方（official），其余用户按请求或默认为 user。
+    if current_user.role == "admin":
+        app_source = "official"
+    elif req.source and req.source in ("official", "user"):
+        app_source = req.source
+    else:
+        app_source = "user"
+
+    app = MarketApp(
+        id=gen_uuid(),
+        name=req.name,
+        description=req.description,
+        icon=req.icon,
+        type=req.type if req.type in ("skill","agent","mcp") else "skill",
+        source=app_source,
+        platform=req.platform,
+        price=req.price,
+        install_count=0,
+        rating_avg=0,
+        review_count=0,
+        publisher_id=current_user.id,
+        publisher_name=current_user.username,
+        is_member_free=req.is_member_free if req.is_member_free is not None else False,
+        join_promotion=req.join_promotion if req.join_promotion is not None else False,
+        agent_endpoint=req.agent_endpoint,
+        agent_api_key=req.api_key,
+        status="published",  # 用户创建的应用直接发布
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+    )
+    db.add(app)
+    db.commit()
+    db.refresh(app)
+
+    # MCP 应用同步注册到后台 MCP Client 管理（mcp_servers 表）
+    if app.type == "mcp":
+        transport = "streamable_http"
+        code = req.download_url or ""
+        if code:
+            app.skill_content = code
+        app.mcp_transport = transport
+        db.commit()
+        db.refresh(app)
+
+        existing_server = db.query(MCPServer).filter(MCPServer.id == app.id).first()
+        if not existing_server:
+            url = _extract_url_from_text(code) or _extract_url_from_text(req.description or "") or ""
+            server = MCPServer(
+                id=app.id,
+                name=app.name,
+                description=app.description or "",
+                transport_type=transport,
+                command="",
+                args="{}",
+                env_vars="{}",
+                url=url,
+                headers="{}",
+                is_enabled=True,
+                status="disconnected",
+                tools_count=0,
+                resources_count=0,
+                created_at=app.created_at or datetime.now(),
+                updated_at=app.updated_at or datetime.now(),
+            )
+            db.add(server)
+            db.commit()
+
+    # 用户创建/导入的应用自动标记为已安装
+    existing_install = db.query(AppInstallation).filter(
+        AppInstallation.user_id == current_user.id,
+        AppInstallation.app_id == app.id,
+    ).first()
+    if not existing_install:
+        installation = AppInstallation(
+            user_id=current_user.id,
+            app_id=app.id,
+            installed_at=datetime.now(),
+        )
+        db.add(installation)
+        db.commit()
+
+    LogService.log_operation(
+        user_id=current_user.id, username=current_user.username, action="upload_app",
+        target="market_app", target_id=app.id,
+        detail=f"上传应用 {app.name}",
+    )
+
+    return {
+        "code": 0,
+        "data": {
+            "id": app.id, "name": app.name, "status": app.status,
+            "created_at": str(app.created_at),
+        },
+        "message": "提交成功，等待审核",
+    }
+
+
+@router.post("/apps/{app_id}/purchase")
+def purchase_app(
+    app_id: str,
+    req: PurchaseAppRequest,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    app = db.query(MarketApp).filter(
+        and_(MarketApp.id == app_id, MarketApp.status == "published")
+    ).first()
+    if not app:
+        raise HTTPException(status_code=404, detail="应用不存在或未上架")
+
+    # 检查是否已购买
+    existing = db.query(AppPurchase).filter(
+        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
+    ).first()
+    if existing and existing.status in ("purchased", "installed"):
+        raise HTTPException(status_code=400, detail="您已购买该应用")
+
+    amount = req.amount or app.price
+
+    # 创建订单
+    order = Order(
+        id=gen_uuid(),
+        user_id=current_user.id,
+        order_type="app",
+        item_id=app_id,
+        item_name=app.name,
+        amount=amount,
+        status="pending",
+        created_at=datetime.now(),
+    )
+    db.add(order)
+    db.flush()
+
+    # 创建购买记录
+    purchase = AppPurchase(
+        id=gen_uuid(),
+        user_id=current_user.id,
+        app_id=app_id,
+        order_id=order.id,
+        status="purchased",
+        purchased_at=datetime.now(),
+    )
+    db.add(purchase)
+
+    # 更新应用下载量
+    app.install_count = (app.install_count or 0) + 1
+    app.updated_at = datetime.now()
+
+    db.commit()
+
+    return {
+        "code": 0,
+        "data": {
+            "order_id": order.id,
+            "purchase_id": purchase.id,
+            "amount": amount,
+            "status": "purchased",
+        },
+        "message": "购买成功",
+    }
+
+
+@router.post("/apps/{app_id}/install")
+def install_app(
+    app_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
+    if not app:
+        raise HTTPException(status_code=404, detail="应用不存在")
+
+    # 检查是否已安装（通过 AppInstallation 表）
+    existing_install = db.query(AppInstallation).filter(
+        and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
+    ).first()
+    if existing_install:
+        raise HTTPException(status_code=400, detail="应用已安装")
+
+    # 检查是否已购买（付费应用需要先购买）
+    purchase = db.query(AppPurchase).filter(
+        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
+    ).first()
+
+    # 对于付费应用，必须已购买才能安装
+    if float(app.price or 0) > 0:
+        if not purchase:
+            raise HTTPException(status_code=403, detail="请先购买应用")
+        if purchase.status == "installed":
+            raise HTTPException(status_code=400, detail="应用已安装")
+
+    # 创建安装记录
+    installation = AppInstallation(
+        id=gen_uuid(),
+        user_id=current_user.id,
+        app_id=app_id,
+        installed_at=datetime.now(),
+    )
+    db.add(installation)
+
+    # 如果有购买记录，更新状态为已安装
+    if purchase:
+        purchase.status = "installed"
+        purchase.installed_at = datetime.now()
+
+    # 更新应用安装量
+    app.install_count = (app.install_count or 0) + 1
+    app.updated_at = datetime.now()
+
+    db.commit()
+
+    return {
+        "code": 0,
+        "data": {"installation_id": installation.id, "status": "installed"},
+        "message": "安装成功",
+    }
+
+
+@router.delete("/apps/{app_id}/uninstall")
+def uninstall_app(
+    app_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """卸载应用 - 删除当前用户的安装记录（按用户隔离）"""
+    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
+    if not app:
+        raise HTTPException(status_code=404, detail="应用不存在")
+
+    # 查找当前用户的安装记录（按用户隔离）
+    installation = db.query(AppInstallation).filter(
+        and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
+    ).first()
+    if not installation:
+        raise HTTPException(status_code=400, detail="应用未安装")
+
+    # 删除安装记录
+    db.delete(installation)
+
+    # 问题4修复: 将AppPurchase状态从installed重置为purchased
+    purchase = db.query(AppPurchase).filter(
+        and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
+    ).first()
+    if purchase and purchase.status == "installed":
+        purchase.status = "purchased"
+
+    # 更新应用安装量（减1，最小为0）
+    if app.install_count and app.install_count > 0:
+        app.install_count = app.install_count - 1
+    app.updated_at = datetime.now()
+
+    db.commit()
+
+    LogService.log_operation(
+        user_id=current_user.id, username=current_user.username, action="uninstall_app",
+        target="market_app", target_id=app_id,
+        detail=f"卸载应用 {app.name}",
+    )
+
+    return {
+        "code": 0,
+        "message": "卸载成功",
+    }
+
+
+@router.post("/apps/{app_id}/review")
+def review_app(
+    app_id: str,
+    req: ReviewAppRequest,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
+    if not app:
+        raise HTTPException(status_code=404, detail="应用不存在")
+
+    # 检查是否已评价
+    existing = db.query(AppReview).filter(
+        and_(AppReview.app_id == app_id, AppReview.user_id == current_user.id)
+    ).first()
+    if existing:
+        # 更新已有评价
+        old_rating = existing.rating
+        existing.rating = req.rating
+        existing.content = req.content
+        existing.updated_at = datetime.now()
+
+        # 重新计算平均分 (总数不变，减去旧评分加新评分)
+        if app.review_count and app.review_count > 0:
+            total_rating = float(app.rating_avg or 0) * app.review_count
+            total_rating = total_rating - old_rating + req.rating
+            app.rating_avg = total_rating / app.review_count
+        else:
+            app.rating_avg = req.rating
+            app.review_count = 1
+    else:
+        # 创建新评价
+        review = AppReview(
+            id=gen_uuid(),
+            app_id=app_id,
+            user_id=current_user.id,
+            rating=req.rating,
+            content=req.content,
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+        )
+        db.add(review)
+
+        # 重新计算平均分
+        app.review_count = (app.review_count or 0) + 1
+        if app.review_count == 1:
+            app.rating_avg = req.rating
+        else:
+            total_rating = float(app.rating_avg or 0) * (app.review_count - 1) + req.rating
+            app.rating_avg = round(total_rating / app.review_count, 1)
+
+    app.updated_at = datetime.now()
+    db.commit()
+
+    return {
+        "code": 0,
+        "data": {
+            "rating_avg": float(app.rating_avg or 0),
+            "review_count": app.review_count or 0,
+        },
+        "message": "评价成功",
+    }
+
+
+@router.get("/my-apps")
+def get_my_apps(
+    status: Optional[str] = Query(None),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    query = db.query(AppPurchase).filter(AppPurchase.user_id == current_user.id)
+    purchases = query.order_by(desc(AppPurchase.created_at)).all()
+
+    result = []
+    for p in purchases:
+        app = db.query(MarketApp).filter(MarketApp.id == p.app_id).first()
+        if app:
+            result.append({
+                "id": p.id,
+                "app_id": app.id,
+                "name": app.name,
+                "icon": app.icon,
+                "type": app.type,
+                "price": float(app.price or 0),
+                "purchased_at": str(p.created_at) if p.created_at else None,
+            })
+
+    return {"code": 0, "data": result, "message": "ok"}
+
+
+@router.get("/reviews/{app_id}")
+def get_reviews(
+    app_id: str,
+    page: int = Query(default=1, ge=1),
+    page_size: int = Query(default=20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    query = db.query(AppReview).filter(AppReview.app_id == app_id)
+    total = query.count()
+
+    reviews = (
+        query.order_by(desc(AppReview.created_at))
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+
+    result = []
+    for r in reviews:
+        user = db.query(User).filter(User.id == r.user_id).first()
+        result.append({
+            "id": r.id,
+            "user_id": r.user_id,
+            "username": user.username if user else "unknown",
+            "avatar": user.avatar if user else None,
+            "rating": r.rating,
+            "content": r.content,
+            "created_at": str(r.created_at) if r.created_at else None,
+        })
+
+    return {
+        "code": 0,
+        "data": {"total": total, "page": page, "page_size": page_size, "items": result},
+        "message": "ok",
+    }
+
+
+# ============================================================
+# Agent部署资源查询接口
+# ============================================================
+
+@router.get("/agent-resources")
+def get_agent_resources(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Agent部署时需要的资源列表：可用的应用、知识库、文档"""
+    # 用户已发布的应用
+    my_apps = db.query(MarketApp).filter(
+        MarketApp.publisher_id == current_user.id,
+        MarketApp.status == "published",
+    ).order_by(desc(MarketApp.updated_at)).all()
+
+    apps_result = []
+    for app in my_apps:
+        apps_result.append({
+            "id": app.id,
+            "name": app.name,
+            "type": app.type,
+            "description": app.description,
+            "install_count": app.install_count or 0,
+        })
+
+    # 用户知识库
+    from ..models import KnowledgeBase, KnowledgeDocument
+    kb_list = db.query(KnowledgeBase).filter(
+        KnowledgeBase.user_id == current_user.id
+    ).order_by(desc(KnowledgeBase.updated_at)).all()
+
+    kb_result = []
+    for kb in kb_list:
+        doc_count = db.query(KnowledgeDocument).filter(
+            KnowledgeDocument.knowledge_base_id == kb.id
+        ).count()
+        kb_result.append({
+            "id": kb.id,
+            "name": kb.name,
+            "description": kb.description,
+            "document_count": doc_count,
+        })
+
+    # 用户已安装的应用（从AppPurchase）
+    installed = db.query(AppPurchase).filter(
+        AppPurchase.user_id == current_user.id
+    ).all()
+    installed_ids = [p.app_id for p in installed]
+    installed_apps = []
+    if installed_ids:
+        iapps = db.query(MarketApp).filter(MarketApp.id.in_(installed_ids)).all()
+        for ia in iapps:
+            installed_apps.append({
+                "id": ia.id,
+                "name": ia.name,
+                "type": ia.type,
+                "description": ia.description,
+            })
+
+    return {
+        "code": 0,
+        "data": {
+            "my_apps": apps_result,
+            "knowledge_bases": kb_result,
+            "installed_apps": installed_apps,
+        },
+        "message": "ok",
+    }
+
diff --git a/frontend/src/pages/front/MarketPage.tsx b/frontend/src/pages/front/MarketPage.tsx
index b726d88..e48e754 100644
--- a/frontend/src/pages/front/MarketPage.tsx
+++ b/frontend/src/pages/front/MarketPage.tsx
@@ -1491,8 +1491,9 @@ const MarketPage: React.FC = () => {
   // Handle hub import - 调用后端 API 创建应用
   const handleHubImport = async (platformName: string, url: string, skillName: string, price?: number, description?: string) => {
     try {
-      const desc = description || `从 ${platformName} 导入的${showHubImport === 'skill' ? 'Skills工具' : 'MCP应用'} (来源: ${url})`;
-      await marketAPI.createApp({
+      const isMcp = showHubImport === 'mcp';
+      const desc = description || `从 ${platformName} 导入的${isMcp ? 'MCP应用' : 'Skills工具'}${!isMcp ? ` (来源: ${url})` : ''}`;
+      const payload: any = {
         name: skillName,
         type: showHubImport || 'skill',
         description: desc,
@@ -1500,7 +1501,11 @@ const MarketPage: React.FC = () => {
         is_member_free: (price || 0) === 0,
         join_promotion: false,
         source: 'user',
-      });
+      };
+      if (isMcp && url) {
+        payload.download_url = url;
+      }
+      await marketAPI.createApp(payload);
       // 重新加载列表
       loadApps();
     } catch (e: any) {