DataLayer MCP Server

MCP Protocol 2025-11-25 · Streamable HTTP · FastMCP + Starlette · Session-scoped dynamic tools

Architecture Overview

Session & Tool Lifecycle

Security Model

Admin REST API

All Admin endpoints require the X-Admin-Secret header, whose secret is derived from XPEPPER — the platform's master encryption key.

Tool Definition Schema

{
  "name": "query_table",
  "title": "Query Table",
  "description": "Execute a SQL query on the connected datasource",
  "url": "http://localhost:8866/fetch",
  "action": "open_table",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query":   { "type": "string",  "description": "SQL query to execute" },
      "limit":   { "type": "integer", "description": "Max rows to return" },
      "schema":  { "type": "string",  "description": "Target schema name" }
    },
    "required": ["query"]
  },
  "annotations": {
    "readOnlyHint": true,
    "destructiveHint": false
  },
  "fixed_params": {
    "connector_id": 42
  }
}

JSON Schema → Python Type Mapping

Optional parameters are wrapped in Optional[T] with default=None. FastMCP reads the signature and generates the JSON Schema automatically.

Dynamic Signature Generation

The server creates full Python functions on the fly — FastMCP has no idea the tools were registered dynamically:

def _make_signature(properties: dict, required: list) -> inspect.Signature:
    params = [inspect.Parameter("ctx", inspect.Parameter.POSITIONAL_OR_KEYWORD)]
    for pname, pdef in properties.items():
        is_required = pname in required
        annotation = _json_type_to_python(pdef.get("type", "string"))
        if not is_required:
            annotation = Optional[annotation]
        params.append(inspect.Parameter(
            pname,
            inspect.Parameter.KEYWORD_ONLY,
            default=inspect.Parameter.empty if is_required else None,
            annotation=annotation,
        ))
    return inspect.Signature(params)

handler.__signature__ is replaced with the generated signature — FastMCP sees (ctx, query: str, limit: Optional[int] = None) instead of (**kwargs).

Middleware: Transparent Namespacing

Multiple sessions can register tools with identical names without any conflicts.

Destructive Operation Confirmation (MCP Elicitation)

When a tool is marked destructiveHint: true and readOnlyHint: false, the server requests explicit user confirmation via the standard MCP Elicitation mechanism:

elicit_result = await ctx.elicit(
    message=f"Confirm execution of '{tool_name}':\n\n{field_value}",
    schema=_ConfirmSchema,   # {"confirmed": bool}
)
if elicit_result.action != "accept" or not elicit_result.data.confirmed:
    return "Command execution cancelled by user."

InMemoryEventStore — Session Resumability

class InMemoryEventStore(EventStore):
    """SSE Last-Event-ID replay for reconnect without losing events."""

    async def store_event(self, stream_id: str, message) -> str:
        event_id = str(uuid.uuid4())
        self._events.append((event_id, stream_id, message))
        return event_id

    async def replay_events_after(self, last_event_id: str, send_callback):
        # Replays all events after last_event_id
        for eid, sid, msg in events_after_last:
            await send_callback(EventMessage(message=msg, event_id=eid))

Limitation: events are stored in process memory. They are lost on server restart. For production reliability, swap in a Redis-backed implementation.

Performance & Concurrency

{
  "title": { "text": "Server Characteristics", "left": "center", "textStyle": { "fontSize": 14 } },
  "radar": {
    "indicator": [
      { "name": "Session\nConcurrency",   "max": 100 },
      { "name": "Tool\nIsolation",        "max": 100 },
      { "name": "Security",               "max": 100 },
      { "name": "MCP\nCompliance",        "max": 100 },
      { "name": "Extensibility",          "max": 100 },
      { "name": "Performance",            "max": 100 }
    ],
    "radius": 110
  },
  "series": [{
    "type": "radar",
    "data": [{
      "value": [95, 100, 88, 100, 92, 85],
      "name": "DataLayer MCP",
      "areaStyle": { "opacity": 0.25 },
      "lineStyle": { "width": 2, "color": "#1976d2" },
      "itemStyle": { "color": "#1976d2" }
    }]
  }]
}

Startup Configuration

# Environment variables
HOST=0.0.0.0      # bind address
PORT=8080         # MCP server port

# In dbwebtool/settings.py
MCP_ADMIN_SECRET   # secret for /admin/* endpoints
DATAPLAYER_API_URL # backend base URL (for AuthSettings)
XPEPPER            # master key (MCP_ADMIN_SECRET is derived from it)

# Run directly
python dataplayer_mcp_server.py

# Or via uvicorn
uvicorn dataplayer_mcp_server:create_app --host 0.0.0.0 --port 8080 --factory

The backend starts the MCP server as a subprocess automatically when MCP_AUTOSTART=True during startup_handler().

Example: Full Registration Cycle via Admin API

# 1. Initialize session
curl -X POST http://localhost:8080/admin/session/init \
  -H "X-Admin-Secret: <secret>" \
  -H "Content-Type: application/json" \
  -d '{"session_id": "sess_42", "user_token": "tok_abc", "user_id": 7}'

# 2. Register tools
curl -X POST http://localhost:8080/admin/tools/register \
  -H "X-Admin-Secret: <secret>" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "sess_42",
    "tools": [{
      "name": "run_query",
      "description": "Run a SQL query",
      "url": "http://localhost:8866/fetch",
      "action": "open_table",
      "inputSchema": {
        "type": "object",
        "properties": { "query": { "type": "string" } },
        "required": ["query"]
      },
      "annotations": { "readOnlyHint": true }
    }]
  }'

# 3. AI client connects to /mcp
# Authorization: Bearer tok_abc
# → sees tool "run_query" (no prefix)

# 4. Cleanup after session ends
curl -X POST http://localhost:8080/admin/session/cleanup \
  -H "X-Admin-Secret: <secret>" \
  -d '{"session_id": "sess_42"}'

Integration with DataPlayer

The LangGraph AI agent: 1. Receives a user request 2. Uses search_tools(semantic_query) to find relevant tools in Qdrant 3. Connects to the MCP server with the session Bearer token 4. Invokes tools — they HTTP-proxy calls to the DataPlayer API 5. Results are streamed back to the client via WebSocket (/ws/)

Key Files