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 发布者 API 的 URL	http://localhost:4000

示例：

# 设置发布者 URL
export MALLOY_PUBLISHER_ROOT_URL="http://malloy-publisher:4000"

# 使用自定义配置运行
python -m malloy_m_服务器

执行查询

服务器提供了一个 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 发布者客户端用于 Malloy 交互
• Pydantic 用于数据验证

关键组件：

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

贡献

1. 分叉仓库
2. 创建功能分支
3. 进行更改
4. 为新功能添加测试
5. 提交拉取请求

许可证

MIT 许可证 - 详情见 LICENSE 文件