AI导哈

Malloy查询服务器

一个用于执行Malloy查询和管理Malloy资源的MCP服务器实现，适用于数据分析和查询处理场景。

  • by namabile
  • Python
  • 本地部署
  • 2 星标
  • 更新于
在 GitHub 查看 namabile/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 马洛伊发布商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

执行查询

服务器提供了一个MCP工具来执行Malloy查询：

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
  1. 安装依赖项：
uv pip install -e ".[dev]"

运行测试

# Run all tests
pytest

# Run with coverage
pytest --cov=malloy_mcp_server

代码质量

该项目使用：

  • black 用于代码格式化
  • mypy 用于类型检查
  • ruff 对于linting

运行质量检查：

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)

建筑

服务器构建于：

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

关键部件：

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

贡献

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

许可证

MIT许可证-有关详细信息，请参阅许可证文件

相关 MCP