用 MCP 让时序数据库开口说话：Amazon QuickSight + KDB-X 实战

用户自然语言提问
    ↓
QuickSight / LLM 解析意图
    ↓
MCP Client 调用 KDB-X MCP Server 的 tools
    ↓
Server 生成 Q 查询 → 发往 KDB-X 执行
    ↓
结果回传 → QuickSight 渲染图表或表格

{
  "mcpServers": {
    "kdbx-timeseries": {
      "command": "python",
      "args": ["-m", "kdbx_mcp_server"],
      "env": {
        "KDBX_HOST": "localhost",
        "KDBX_PORT": "5000",
        "KDBX_TABLES": "trade,quote,sensor_log",
        "KDBX_DEFAULT_SCHEMA": "market"
      }
    }
  }
}

# kdbx_mcp_server.py — 保存后用 python -m kdbx_mcp_server 启动
import os
from mcp.server import Server
from mcp.types import Tool, TextContent
from qpython import qconnection

# 连接 KDB-X / kdb+
q_conn = qconnection.QConnection(
    host=os.environ.get("KDBX_HOST", "localhost"),
    port=int(os.environ.get("KDBX_PORT", "5000")),
)
q_conn.open()

server = Server("kdbx-timeseries")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="list_tables",
            description="列出 KDB-X 中所有可用的时序表",
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="query_timeseries",
            description="对指定时序表执行聚合查询，支持时间范围过滤和分组",
            inputSchema={
                "type": "object",
                "properties": {
                    "table": {"type": "string", "description": "表名，如 trade"},
                    "start_time": {"type": "string", "description": "起始时间，ISO 格式"},
                    "end_time": {"type": "string", "description": "结束时间，ISO 格式"},
                    "agg": {"type": "string", "description": "聚合函数，如 max、avg、count"},
                    "group_by": {"type": "string", "description": "分组字段，如 sym、date"},
                    "where": {"type": "string", "description": "额外过滤条件"},
                },
                "required": ["table", "start_time", "end_time"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name, arguments):
    if name == "list_tables":
        tables = os.environ.get("KDBX_TABLES", "").split(",")
        return [TextContent(type="text", text=f"可用表: {', '.join(tables)}")]

    if name == "query_timeseries":
        table = arguments["table"]
        start = arguments["start_time"]
        end = arguments["end_time"]
        agg = arguments.get("agg", "avg")
        group_by = arguments.get("group_by", "")
        where = arguments.get("where", "")

        # 构建 Q 查询 — 这里用模板拼接，生产环境应做参数校验防注入
        q_template = f"select {agg} price by {group_by} from {table} where time within(\"{start}\";\"{end}\")"
        if where:
            q_template += f", {where}"

        result = q_conn(q_template)
        # qpython 返回的结果转成可读文本
        return [TextContent(type="text", text=str(result))]

    return [TextContent(type="text", text="未知工具")]

if __name__ == "__main__":
    server.run()

# 假设 MCP Server 已部署到 ECS/Fargate，暴露 HTTPS 终端节点
aws quicksight create-data-source \
  --aws-account-id 123456789012 \
  --data-source-id kdbx-mcp-source \
  --data-source '{"Type":"CUSTOM","CustomDataSourceConfig":{"Url":"https://kdbx-mcp.your-vpc.elb.amazonaws.com/mcp"}}'

# 然后在 QuickSight Q 中将此数据源关联到分析面板
aws quicksight update-analysis \
  --aws-account-id 123456789012 \
  --analysis-id market-analysis \
  --data-source-arns arn:aws:quicksight:us-east-1:123456789012:data-source/kdbx-mcp-source