Malloy MCP 服务器是一个用于执行 Malloy 查询和管理 Malloy 资源的 MCP 服务器实现,能有效助力用户执行查询和管理资源。
# 使用 uv 安装(推荐)
uv pip install malloy-mcp-server
# 或使用 pip
pip install malloy-mcp-server
from malloy_mcp_server import mcp
# 运行服务器
if __name__ == "__main__":
mcp.serve()
服务器可以通过环境变量进行配置:
| 变量 | 描述 | 默认值 |
|---|---|---|
MALLOY_PUBLISHER_ROOT_URL |
Malloy 发布者 API 的 URL | http://localhost:4000 |
示例:
# 设置发布者 URL
export MALLOY_PUBLISHER_ROOT_URL="http://malloy-publisher:4000"
# 使用自定义配置运行
python -m malloy_mcp_server
服务器提供了一个 MCP 工具用于执行 Malloy 查询:
from malloy_mcp_server import ExecuteMalloyQueryTool
# 示例查询执行
result = await ExecuteMalloyQueryTool(
query="select * from users",
model_path="my_package/users"
)
服务器提供了以下资源端点:
malloy://project/home/metadata - 项目元数据malloy://project/home/package/{package_name} - 包元数据malloy://project/home/model/{model_path} - 模型元数据git clone https://github.com/namabile/malloy-mcp-server.git
cd malloy-mcp-server
uv pip install -e ".[dev]"
# 运行所有测试
pytest
# 带覆盖率运行
pytest --cov=malloy_mcp_server
项目使用:
black 进行代码格式化mypy 进行类型检查ruff 进行代码检查运行质量检查:
black .
mypy .
ruff check .
服务器提供详细的错误处理和上下文:
from malloy_mcp_server.errors import QueryExecutionError
try:
result = await ExecuteMalloyQueryTool(...)
except QueryExecutionError as e:
print(f"错误: {e.message}")
print("上下文:", e.context)
服务器基于以下构建:
关键组件:
server.py - 核心服务器实现tools/query_executor.py - 查询执行工具errors.py - 错误处理实用程序本项目采用 MIT 许可证,详见 LICENSE 文件。