Malloy MCP Server

🚀 Malloy MCP Server

An MCP server implementation for executing Malloy queries and managing Malloy resources.

✨ Features

• Execute Malloy queries via MCP.
• Access metadata of Malloy projects, packages, and models.
• Powerful error handling with detailed context.
• Comprehensive test coverage.
• Type-safe implementation.

📦 Installation

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

# Or use pip
pip install malloy-mcp-server

💻 Usage Examples

Basic Usage

Start the Server

from malloy_mcp_server import mcp

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

Configuration

The server can be configured via environment variables:

Example:

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

# Run with custom configuration
python -m malloy_mcp_server

Execute a Query

The server provides an MCP tool for executing Malloy queries:

from malloy_mcp_server import ExecuteMalloyQueryTool

# Example query execution
result = await ExecuteMalloyQueryTool(
    query="select * from users",
    model_path="my_package/users"
)

Access Resources

The server provides the following resource endpoints:

• malloy://project/home/metadata - Project metadata
• malloy://project/home/package/{package_name} - Package metadata
• malloy://project/home/model/{model_path} - Model metadata

Advanced Usage

The above examples cover the basic usage scenarios. For more complex operations, you can refer to the source code and expand the functionality according to your needs.

🔧 Technical Details

Setup

1. Clone the repository:

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

2. Install dependencies:

uv pip install -e ".[dev]"

Run Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=malloy_mcp_server

Code Quality

The project uses:

• black for code formatting.
• mypy for type checking.
• ruff for code inspection.

Run quality checks:

black .
mypy .
ruff check .

Error Handling

The server provides detailed error handling and context:

from malloy_mcp_server.errors import QueryExecutionError

try:
    result = await ExecuteMalloyQueryTool(...)
except QueryExecutionError as e:
    print(f"Error: {e.message}")
    print("Context:", e.context)

Architecture

The server is built on the following:

• FastMCP for MCP server implementation.
• Malloy publisher client for Malloy interaction.
• Pydantic for data validation.

Key components:

• server.py - Core server implementation.
• tools/query_executor.py - Query execution tool.
• errors.py - Error handling utilities.

🤝 Contribution

1. Fork the repository.
2. Create a feature branch.
3. Make modifications.
4. Add tests for new features.
5. Submit a pull request.

📄 License

MIT License - See the LICENSE file for details.