Malloy MCP 服务器

一个用于执行 Malloy 查询和管理 Malloy 资源的 MCP 服务器实现。

特性

• 通过 MCP 执行 Malloy 查询
• 访问 Malloy 项目、包和模型元数据
• 强大的错误处理，提供详细的上下文
• 全面的测试覆盖
• 类型安全实现

安装

# 使用 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 Publisher API 的 URL	http://localhost:4000

示例：

# 设置 publisher 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} - 模型元数据

开发

设置

1. 克隆仓库：

git clone https://github.com/namabile/malloy-mcp-server.git
cd malloy-mcp-server

2. 安装依赖：

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)

架构

服务器构建于：

• FastMCP，用于 MCP 服务器实现
• Malloy Publisher Client，用于 Malloy 交互
• Pydantic，用于数据验证

关键组件：

• server.py - 核心服务器实现
• tools/query_executor.py - 查询执行工具
• errors.py - 错误处理工具

贡献

1. Fork 仓库
2. 创建一个特性分支
3. 进行更改
4. 为新功能添加测试
5. 提交 Pull Request

许可证

MIT 许可证 - 详细信息请参阅 LICENSE 文件