Merge branch 'feature/mcp-integration' into develop

Author: ewlarson

.env.example

@@ -46,7 +46,8 @@ OLD_DB_NAME=geoportal_production_20251030
 # =============================================================================
 
 POSTGRES_PASSWORD="<secret>"
-DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:2345/btaa_geospatial_api
+# IMPORTANT: keep the password in DATABASE_URL in sync with POSTGRES_PASSWORD.
+DATABASE_URL=postgresql+asyncpg://postgres:<secret>@localhost:2345/btaa_geospatial_api
 
 # =============================================================================
 # Elasticsearch

README.md

@@ -129,6 +129,18 @@ In addition to the website and API, this repository contains a **QGIS Plugin** t
 - **Source code**: Located in the `qgis-plugin/` directory.
 - **Testing & Development**: See `qgis-plugin/docs/testing.md` for instructions on running tests and linting.
 
+## Claude Desktop / MCP
+
+This repository now includes the same MCP bridge layer that existed in the older `ogm-api` project:
+
+- `mcp/run_mcp_service.py` runs the stdio MCP server from the repo root
+- `mcp/mcp_http_bridge.js` forwards stdio MCP traffic to `POST /api/v1/mcp`
+- `mcp/mcp_websocket_bridge.js` forwards stdio MCP traffic to `/api/v1/mcp/ws`
+- `mcp/run_mcp_websocket_bridge.py` launches the WebSocket bridge with Node 18+ automatically
+- `mcp/claude_mcp_config.json` is a Claude Desktop template (`cwd` = your clone root; see `docs/mcp/`)
+
+More detail is in `docs/mcp/README.md` and `docs/mcp/claude_desktop.md`.
+
 ## Documentation (for staff who want details)
 
 All documentation is now in the top-level `docs/` folder:
@@ -137,6 +149,7 @@ All documentation is now in the top-level `docs/` folder:
 - **Search**: `docs/backend/search.md`
 - **Service tiers / API keys / rate limiting**: `docs/backend/service_tiers_runbook.md`
 - **Scripts (Python utilities)**: `docs/backend/scripts.md`
+- **MCP / Claude Desktop**: `docs/mcp/`
 - **Frontend docs**: `docs/frontend/`
 - **QGIS plugin docs**: `qgis-plugin/docs/`
 - **Developer Make tasks**: `docs/make_tasks.md`

backend/app/api/v1/endpoint_modules/mcp.py

@@ -1,9 +1,9 @@
 import logging
 
-from fastapi import APIRouter, WebSocket, WebSocketDisconnect
-from fastapi.responses import JSONResponse
+from fastapi import APIRouter, Request, WebSocket, WebSocketDisconnect
+from fastapi.responses import JSONResponse, Response
 
-from app.services.mcp_service import get_mcp_service_info
+from app.services.mcp_service import get_mcp_service_info, handle_mcp_message
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
@@ -15,6 +15,27 @@ async def mcp_endpoint():
     return JSONResponse(content=get_mcp_service_info())
 
 
+@router.post("/mcp")
+async def mcp_http_transport(request: Request):
+    """Handle JSON-RPC-over-HTTP requests for MCP bridge compatibility."""
+    try:
+        payload = await request.json()
+    except Exception:
+        return JSONResponse(
+            content={
+                "jsonrpc": "2.0",
+                "id": None,
+                "error": {"code": -32700, "message": "Parse error"},
+            }
+        )
+
+    response = await handle_mcp_message(payload)
+    if response is None:
+        return Response(status_code=204)
+
+    return JSONResponse(content=response)
+
+
 @router.websocket("/mcp/ws")
 async def mcp_websocket_endpoint(websocket: WebSocket):
     """WebSocket endpoint for MCP service."""

backend/app/services/mcp_service.py

@@ -360,64 +360,62 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> CallToolResu
 
     async def _search_resources(self, arguments: Dict[str, Any]) -> CallToolResult:
         """Search for resources."""
+        query = arguments.get("query")
+        page = arguments.get("page", 1)
+        per_page = arguments.get("per_page", 10)
         try:
-            query = arguments.get("query")
-            page = arguments.get("page", 1)
-            per_page = arguments.get("per_page", 10)
             sort = arguments.get("sort")
-
-            search_service = SearchService()
-            results = await search_service.search(
-                q=query,
-                page=page,
-                limit=per_page,
-                sort=sort,
-                request_query_params="",
-                callback=None,
+            params = {
+                "q": query,
+                "page": page,
+                "per_page": per_page,
+                "sort": sort,
+            }
+            payload = await self._api_request(
+                "/search", params={key: value for key, value in params.items() if value is not None}
             )
-
-            # Process each resource to get full details
-            processed_resources = []
-            async with get_async_session()() as session:
-                for item in results.get("data", []):
-                    try:
-                        # Extract the resource data from the search result
-                        resource_dict = item.get("attributes", {})
-                        if not resource_dict:
-                            continue
-
-                        # Process the resource using the same logic as API endpoints
-                        from app.api.v1.utils import process_resource
-
-                        resource_object = await process_resource(resource_dict, session)
-                        processed_resources.append(resource_object)
-                    except Exception as e:
-                        logger.error(f"Error processing search result: {str(e)}", exc_info=True)
-                        continue
-
-            # Return the full resource objects as JSON
-            return CallToolResult(
-                content=[
-                    TextContent(
-                        type="text",
-                        text=json.dumps(
-                            {
-                                "query": query,
-                                "total_results": len(results.get("data", [])),
-                                "page": page,
-                                "per_page": per_page,
-                                "resources": processed_resources,
-                            },
-                            indent=2,
-                        ),
-                    )
-                ]
+            response_data = payload.get("data")
+            response_meta = response_data.get("meta", {}) if isinstance(response_data, dict) else {}
+            resources = response_data.get("data", []) if isinstance(response_data, dict) else []
+
+            if payload["status_code"] >= 400:
+                payload["query"] = query
+                return await self._tool_result_json(payload, is_error=True)
+
+            return await self._tool_result_json(
+                {
+                    "query": query,
+                    "total_results": response_meta.get("totalCount", len(resources)),
+                    "total_pages": response_meta.get("totalPages"),
+                    "page": response_meta.get("currentPage", page),
+                    "per_page": response_meta.get("perPage", per_page),
+                    "spelling_suggestions": response_meta.get("spellingSuggestions", []),
+                    "resources": resources,
+                    "links": (
+                        response_data.get("links", {}) if isinstance(response_data, dict) else {}
+                    ),
+                    "source": {
+                        "transport": "http",
+                        "url": payload.get("url"),
+                        "status_code": payload.get("status_code"),
+                    },
+                }
             )
         except Exception as e:
             logger.error(f"Error in _search_resources: {e}", exc_info=True)
-            return CallToolResult(
-                content=[TextContent(type="text", text=f"Error searching resources: {str(e)}")],
-                isError=True,
+            return await self._tool_result_json(
+                {
+                    "error": "Search request failed",
+                    "detail": str(e),
+                    "query": query,
+                    "page": page,
+                    "per_page": per_page,
+                    "source": {
+                        "transport": "http",
+                        "url": f"{self._public_api_base()}/api/v1/search",
+                    },
+                },
+                is_error=True,
             )
 
     async def _get_resource(self, arguments: Dict[str, Any]) -> CallToolResult:
@@ -622,7 +620,7 @@ async def _get_resource_viewer(self, arguments: Dict[str, Any]) -> CallToolResul
             embed = arguments.get("embed", False)
 
             # Build the record URL for the viewer
-            base_url = "http://localhost:8000"
+            base_url = self._public_api_base()
             record_url = f"{base_url}/api/v1/resources/{resource_id}/metadata"
 
             # Create the HTML content
@@ -677,7 +675,10 @@ async def _get_resource_viewer(self, arguments: Dict[str, Any]) -> CallToolResul
 
     def _public_api_base(self) -> str:
         """Resolve base URL for calling this API over HTTP."""
-        base = os.getenv("APPLICATION_URL", "http://localhost:8000").rstrip("/")
+        base = os.getenv("BTAA_GEOSPATIAL_API_BASE_URL") or os.getenv(
+            "APPLICATION_URL", "http://localhost:8000"
+        )
+        base = base.rstrip("/")
         if base.endswith("/api/v1"):
             base = base[: -len("/api/v1")]
         return base
@@ -967,7 +968,8 @@ async def run_mcp_websocket_server(websocket):
             try:
                 data = json.loads(message)
                 response = await handle_mcp_message(data)
-                await websocket.send_text(json.dumps(response))
+                if response is not None:
+                    await websocket.send_text(json.dumps(response))
             except json.JSONDecodeError:
                 await websocket.send_text(
                     json.dumps(
@@ -987,12 +989,30 @@ async def run_mcp_websocket_server(websocket):
         logging.error(f"WebSocket error: {e}")
 
 
-async def handle_mcp_message(data: Dict[str, Any]) -> Dict[str, Any]:
+async def handle_mcp_message(data: Dict[str, Any]) -> Dict[str, Any] | None:
     """Handle MCP protocol messages."""
+    if not isinstance(data, dict):
+        return {
+            "jsonrpc": "2.0",
+            "id": None,
+            "error": {"code": -32600, "message": "Invalid request"},
+        }
+
     method = data.get("method")
     msg_id = data.get("id")
     params = data.get("params", {})
 
+    if not method or not isinstance(method, str):
+        return {
+            "jsonrpc": "2.0",
+            "id": msg_id,
+            "error": {"code": -32600, "message": "Invalid request"},
+        }
+
+    # Ignore notifications that do not expect a reply.
+    if method in {"notifications/initialized", "$/cancelRequest"} and msg_id is None:
+        return None
+
     if method == "initialize":
         return {
             "jsonrpc": "2.0",
@@ -1004,6 +1024,9 @@ async def handle_mcp_message(data: Dict[str, Any]) -> Dict[str, Any]:
             },
         }
 
+    elif method == "ping":
+        return {"jsonrpc": "2.0", "id": msg_id, "result": {}}
+
     elif method == "tools/list":
         return {"jsonrpc": "2.0", "id": msg_id, "result": {"tools": mcp_service.tool_specs}}
 
@@ -1085,8 +1108,8 @@ def get_mcp_service_info() -> dict[str, Any]:
         "connections": {
             "stdio": {
                 "type": "stdio",
-                "command": "python",
-                "args": ["-m", "app.services.mcp_service"],
+                "command": "python3",
+                "args": ["mcp/run_mcp_service.py"],
             },
             "websocket": {"type": "websocket", "url": "/api/v1/mcp/ws"},
         },

backend/db/config.py

@@ -1,5 +1,5 @@
 import os
-from urllib.parse import urlparse, urlunparse
+from urllib.parse import quote, urlparse, urlunparse
 
 from dotenv import load_dotenv
 
@@ -10,12 +10,33 @@
     # In sandboxed environments, .env may be unreadable. Continue with defaults/env.
     pass
 
+def _repair_placeholder_database_password(database_url: str | None) -> str | None:
+    """Replace placeholder `postgres` password with the configured env password."""
+    if not database_url:
+        return database_url
+
+    parsed = urlparse(database_url)
+    env_password = os.getenv("POSTGRES_PASSWORD") or os.getenv("DB_PASSWORD")
+
+    # Only repair the common local-dev placeholder case so we do not unexpectedly
+    # rewrite intentionally different DATABASE_URL values.
+    if not env_password or parsed.password != "postgres" or env_password == "postgres":
+        return database_url
+
+    username = parsed.username or ""
+    password = quote(env_password, safe="")
+    hostname = parsed.hostname or ""
+    port = f":{parsed.port}" if parsed.port else ""
+    netloc = f"{username}:{password}@{hostname}{port}" if username else f"{hostname}{port}"
+    return urlunparse(parsed._replace(netloc=netloc))
+
+
 # Get database configuration from environment variables
 # Use DATABASE_URL if provided, otherwise construct from individual components
 DATABASE_URL = os.getenv("DATABASE_URL")
 if not DATABASE_URL:
     DB_USER = os.getenv("DB_USER", "postgres")
-    DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")
+    DB_PASSWORD = os.getenv("POSTGRES_PASSWORD") or os.getenv("DB_PASSWORD", "postgres")
     # Default to localhost when running outside Docker (e.g., tests)
     # Use paradedb (Docker service name) when running inside Docker
     is_docker = os.getenv("IS_DOCKER") == "true"
@@ -27,6 +48,8 @@
     # Construct database URL with asyncpg driver, always targeting btaa_geospatial_api
     DATABASE_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
 else:
+    DATABASE_URL = _repair_placeholder_database_password(DATABASE_URL)
+
     # Convert Docker hostnames to localhost when running outside Docker
     # This handles cases where DATABASE_URL is set from .env with Docker service names
     is_docker = os.getenv("IS_DOCKER") == "true"

backend/pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "tomli==2.2.1",
     "typing_extensions>=4.14.1",
     "uvicorn==0.32.1",
+    "websockets==15.0.1",
     "requests==2.33.0",
     "docker==7.1.0",
     "elasticsearch>=9.0.0",
@@ -130,4 +131,3 @@ directory = "coverage_html_report"
 pythonpath = ["."]
 testpaths = ["tests"]
 python_files = ["test_*.py"]
-

backend/scripts/start_web_singlehost.sh

@@ -7,7 +7,7 @@ export PATH="/opt/venv/bin:$PATH"
 
 echo "[start_web_singlehost] starting FastAPI (uvicorn) on 127.0.0.1:8001"
 cd /app/backend
-python -m uvicorn app.main:app --host 127.0.0.1 --port 8001 &
+python -m uvicorn app.main:app --host 127.0.0.1 --port 8001 --ws websockets &
 UVICORN_PID=$!
 
 echo "[start_web_singlehost] starting SSR (react-router-serve) on 0.0.0.0:3000"
@@ -30,4 +30,3 @@ trap cleanup EXIT INT TERM
 
 echo "[start_web_singlehost] starting nginx (public) on 0.0.0.0:8000"
 exec nginx -g 'daemon off;'
-