使用说明

EdgeLake MCP服务器

用于EdgeLake分布式数据库的模型上下文协议（MCP）服务器，为AI助手提供跨EdgeLake节点查询和探索分布式数据的访问权限。

特性

资源发现：列出EdgeLake节点上可用的所有数据库和表
架构检查：使用列信息检索表架构
SQL查询执行：使用以下命令执行复杂的SQL查询：
- 带有AND/OR运算符的WHERE子句
- 按聚合分组
- 按ASC/DESC排序
- 跨表和数据库的JOIN
- 扩展元数据字段（+ip、+hostname、@table_name等）
- 结果分页限制
多线程执行：并发请求处理以实现最佳性能
无状态设计：不需要会话管理

建筑

┌────────────────────┐         ┌─────────────────────────┐         ┌────────────────────┐
│                    │         │   EdgeLake MCP Server   │         │                    │
│    MCP Client      │◀───────▶│                         │◀───────▶│   EdgeLake Node    │
│  (Claude, etc.)    │  stdio  │   - Resources (list)    │  HTTP   │   (REST API)       │
│                    │         │   - Resources (read)    │         │                    │
└────────────────────┘         │   - Tools (query)       │         └────────────────────┘
                               │   - Tools (node_status) │
                               └─────────────────────────┘

安装

先决条件

Python 3.10或更高版本
访问启用REST API的EdgeLake节点
EdgeLake节点在可访问的IP:port上运行（默认值：localhost:32049）

再进行

pip install -r requirements.txt

配置

使用环境变量配置服务器：

一切： 更新以动态提供节点信息

变量	描述	默认值
`EDGELAKE_HOST`	EdgeLake节点IP/主机名	`127.0.0.1`
`EDGELAKE_PORT`	EdgeLake REST API端口	`32049`
`EDGELAKE_TIMEOUT`	HTTP请求超时（秒）	`20`
`EDGELAKE_MAX_WORKERS`	最大并发线程数	`10`
`LOG_LEVEL`	日志记录级别（调试、信息、警告、错误）	`INFO`

配置示例

创建一个 .env 文件：

EDGELAKE_HOST=192.168.1.106
EDGELAKE_PORT=32049
EDGELAKE_TIMEOUT=30
EDGELAKE_MAX_WORKERS=20
LOG_LEVEL=INFO

或导出环境变量：

export EDGELAKE_HOST=192.168.1.106
export EDGELAKE_PORT=32049

用法

运行服务器

MCP服务器使用stdio传输作为子进程运行：

python server.py

MCP客户端配置

添加到您的MCP客户端配置中（例如，Claude Desktop）：

macOS/Linux： `~/Library/Application Support/Claude/claude_desktop_config.json`

窗户： `%APPDATA%\Claude\claude_desktop_config.json`

{
  "mcpServers": {
    "edgelake": {
      "command": "python",
      "args": ["/path/to/edgelake/mcp-server/server.py"],
      "env": {
        "EDGELAKE_HOST": "192.168.1.106",
        "EDGELAKE_PORT": "32049"
      }
    }
  }
}

MCP协议实现

资源

`resources/list`

列出所有可用的数据库和表。

响应格式：

database://{database_name}                     - Database resource
database://{database_name}/{table_name}        - Table resource

例子：

[
  {
    "uri": "database://my_database",
    "name": "Database: my_database",
    "description": "All tables in database 'my_database'",
    "mimeType": "application/json"
  },
  {
    "uri": "database://my_database/users",
    "name": "my_database.users",
    "description": "Table 'users' in database 'my_database'",
    "mimeType": "application/json"
  }
]

`resources/read`

读取特定资源（表架构）。

URI格式： database://{database}/{table}

请求示例：

{
  "method": "resources/read",
  "params": {
    "uri": "database://my_database/users"
  }
}

示例响应：

{
  "contents": [
    {
      "uri": "database://my_database/users",
      "mimeType": "application/json",
      "text": "{\n  \"columns\": [\n    {\"name\": \"id\", \"type\": \"INTEGER\"},\n    {\"name\": \"name\", \"type\": \"VARCHAR\"},\n    {\"name\": \"email\", \"type\": \"VARCHAR\"}\n  ]\n}"
    }
  ]
}

工具

`query`

使用高级过滤和聚合对EdgeLake执行SQL查询。

参数：

参数	类型	必填	说明
`database`	string	是	数据库名称
`table`	string	是	表名
`select`	array[string]	否	要选择的列（默认值： `["*"]`)
`where`	string	否	WHERE子句条件
`group_by`	array[string]	否	分组依据的列
`order_by`	array[object]	否	排序规范
`include_tables`	array[string]	否	要JOIN的其他表
`extend_fields`	array[string]	否	要添加的元数据字段
`limit`	integer	否	要返回的最大行数（默认值：100）
`format`	string	否	输出格式： `json` 或 `table` （默认值： `json`)

示例-简单查询：

{
  "name": "query",
  "arguments": {
    "database": "iot_data",
    "table": "sensor_readings",
    "where": "temperature > 25",
    "limit": 10
  }
}

示例-复杂聚合：

{
  "name": "query",
  "arguments": {
    "database": "iot_data",
    "table": "sensor_readings",
    "select": ["device_id", "AVG(temperature) as avg_temp", "COUNT(*) as count"],
    "where": "timestamp > '2025-01-01'",
    "group_by": ["device_id"],
    "order_by": [
      {"column": "avg_temp", "direction": "DESC"}
    ],
    "limit": 20
  }
}

示例-跨数据库连接：

{
  "name": "query",
  "arguments": {
    "database": "sales",
    "table": "orders",
    "include_tables": ["inventory.products", "customers"],
    "where": "orders.status = 'completed'",
    "limit": 50
  }
}

示例-扩展字段：

{
  "name": "query",
  "arguments": {
    "database": "iot_data",
    "table": "events",
    "extend_fields": ["+ip", "+hostname", "@table_name"],
    "limit": 100
  }
}

`node_status`

获取EdgeLake节点状态和运行状况信息。

例子：

{
  "name": "node_status",
  "arguments": {}
}

`list_databases`

列出EdgeLake中的所有可用数据库。在查询之前，使用此功能可以发现哪些数据库可用。

例子：

{
  "name": "list_databases",
  "arguments": {}
}

答复：

{
  "databases": ["new_company", "iot_data", "sales"],
  "count": 3
}

`list_tables`

列出特定数据库中的所有表。使用此功能可以在查询之前发现数据库中可用的表。

参数：

参数	类型	必填	说明
`database`	string	是	要列出表的数据库名称

例子：

{
  "name": "list_tables",
  "arguments": {
    "database": "new_company"
  }
}

答复：

{
  "database": "new_company",
  "tables": ["rand_data", "ping_sensor", "events"],
  "count": 3
}

`get_schema`

获取特定表的架构（列定义）。在查询之前，使用此功能了解哪些列可用。

参数：

参数	类型	必填	说明
`database`	string	是	数据库名称
`table`	string	是	表名

例子：

{
  "name": "get_schema",
  "arguments": {
    "database": "new_company",
    "table": "rand_data"
  }
}

答复：

{
  "columns": [
    {"name": "row_id", "type": "SERIAL"},
    {"name": "insert_timestamp", "type": "TIMESTAMP"},
    {"name": "tsd_name", "type": "CHAR(3)"},
    {"name": "tsd_id", "type": "INT"},
    {"name": "timestamp", "type": "timestamp"},
    {"name": "value", "type": "decimal"}
  ]
}

`server_info`

获取EdgeLake MCP服务器版本和配置信息。

例子：

{
  "name": "server_info",
  "arguments": {}
}

答复：

{
  "version": "1.0.6",
  "server_name": "edgelake-mcp-server",
  "configuration": {
    "edgelake_host": "192.168.1.106",
    "edgelake_port": 32349,
    "request_timeout": 20,
    "max_workers": 10,
    "log_level": "INFO"
  }
}

查询构建规则

WHERE子句

使用AND/OR运算符添加过滤条件：

WHERE is_active = true AND age > 18
WHERE status = 'active' OR status = 'pending'
WHERE (category = 'A' OR category = 'B') AND price > 100

分组依据

按列对结果进行分组（使用具有非聚合列的聚合时需要）：

SELECT device_id, AVG(temperature) FROM sensors GROUP BY device_id

排序依据

按列排序结果，方向可选：

ORDER BY created_at DESC
ORDER BY category ASC, price DESC

包括表格（JOIN）

使用逗号分隔语法包含其他表。对于跨数据库表，请使用 db_name.table_name:

FROM orders, customers, inventory.products

扩展字段

使用特殊前缀添加EdgeLake元数据字段：

+ip -节点IP地址
+overlay_ip -覆盖网络IP
+hostname -节点主机名
@table_name -源表名称

SELECT +ip, +hostname, @table_name, * FROM events

限制

限制返回的行数：

SELECT * FROM users LIMIT 100

API示例

使用curl（用于测试）

# List resources
echo '{"jsonrpc":"2.0","id":1,"method":"resources/list","params":{}}' | python server.py

# Read table schema
echo '{"jsonrpc":"2.0","id":2,"method":"resources/read","params":{"uri":"database://mydb/users"}}' | python server.py

# Execute query
echo '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"query","arguments":{"database":"mydb","table":"users","where":"is_active = true","limit":10}}}' | python server.py

# Get node status
echo '{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"node_status","arguments":{}}}' | python server.py

EdgeLake命令参考

MCP服务器通过以下命令使用EdgeLake的REST API：

MCP操作	EdgeLake命令
列出数据库	`GET /` 带标题 `command: get databases`
列出表格	`GET /` 带标题 `command: get tables where dbms = {database}`
获取架构	`GET /` 带标题 `command: get columns where dbms = {database} and table = {table}`
执行查询	`GET /` 带标题 `command: sql {database} format = {format} "{query}"`
节点状态	`GET /` 带标题 `command: get status`

日志记录

日志将写入：

文件: edgelake_mcp.log （在服务器目录中）
标准错误：用于调试的控制台输出

通过设置日志级别 LOG_LEVEL 环境变量（调试、信息、警告、错误）。

发展

项目结构

mcp-server/
├── server.py              # Main MCP server implementation
├── edgelake_client.py     # Multi-threaded EdgeLake HTTP client
├── query_builder.py       # SQL query construction
├── config.py              # Configuration management
├── requirements.txt       # Python dependencies
├── README.md              # This file
└── Design/                # Design documentation
    ├── mcp_service.md
    └── top-level-diagram.monojson

运行测试

pytest

代码的风格

# Format code
black *.py

# Type checking
mypy *.py

故障排除

连接问题

问题：无法连接到EdgeLake节点

Error: Request error: Connection refused

解决方案:

验证EdgeLake节点是否正在运行： curl http://{host}:{port}
检查防火墙设置
验证 EDGELAKE_HOST 和 EDGELAKE_PORT 是正确的

空数据库列表

问题：未从返回数据库 resources/list

解决方案:

检查EdgeLake节点是否有数据库： curl -H "command: get databases" http://{host}:{port}
验证用户是否有查看数据库的权限
检查EdgeLake日志是否有错误

查询超时

问题：查询时间过长，超时

解决方案:

增加 EDGELAKE_TIMEOUT 环境变量
添加更具体的WHERE子句以减少结果集
使用LIMIT限制返回的行

许可证

Mozilla公共许可证2.0

支持

对于问题和疑问：

EdgeLake文件：https://edgelake.github.io
EdgeLake GitHub：https://github.com/EdgeLake
MCP规范：https://modelcontextprotocol.io

贡献

欢迎投稿！请确保：

代码遵循PEP 8风格指南
所有测试均通过
新功能包括文档
类型提示贯穿始终

使用说明

EdgeLake MCP服务器

特性

建筑

安装

先决条件

再进行

配置

配置示例

用法

运行服务器

MCP客户端配置

macOS/Linux： ~/Library/Application Support/Claude/claude_desktop_config.json

窗户： %APPDATA%\Claude\claude_desktop_config.json

MCP协议实现

资源

resources/list

resources/read

工具

query

node_status

list_databases

list_tables

get_schema

server_info

查询构建规则

WHERE子句

分组依据

排序依据

包括表格（JOIN）

扩展字段

限制

API示例

使用curl（用于测试）

EdgeLake命令参考

日志记录

发展

项目结构

运行测试

代码的风格

故障排除

连接问题

空数据库列表

查询超时

许可证

支持

贡献

相关 MCP

DuckDB数据库交互服务

数据库MCP工具箱

MySQL数据库操作工具

Firebird数据库MCP服务器

区块链数据查询服务

IoTDB数据库MCP服务

macOS/Linux： `~/Library/Application Support/Claude/claude_desktop_config.json`

窗户： `%APPDATA%\Claude\claude_desktop_config.json`

`resources/list`

`resources/read`

`query`

`node_status`

`list_databases`

`list_tables`

`get_schema`

`server_info`