Malloy MCP Server

Malloy MCP 服务

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

功能

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

安装

# Install using uv (recommended)
uv pip install malloy-mcp-server

# Or using pip
pip install malloy-mcp-server

使用方法

启动服务

from malloy_mcp_server import mcp

# Run the server
if __name__ == "__main__":
    mcp.serve()

配置

可以使用环境变量配置服务：

变量	描述	默认值
MALLOY_PUBLISHER_ROOT_URL	Malloy Publisher API 的 URL	http://localhost:4000

示例：

# Set the publisher URL
export MALLOY_PUBLISHER_ROOT_URL="http://malloy-publisher:4000"

# Run with custom configuration
python -m malloy_mcp_server

执行查询

服务提供了用于执行 Malloy 查询的 MCP 工具：

from malloy_mcp_server import ExecuteMalloyQueryTool

# Example query execution
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]"

运行测试

# Run all tests
pytest

# Run with coverage
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"Error: {e.message}")
    print("Context:", e.context)

架构

服务基于以下组件构建：

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

主要组件：

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

贡献

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

许可证

MIT 许可证 - 详情请参阅 LICENSE 文件