MCP/mcp-mailu
1
0
Fork 0
forked from rsp2k/mcp-mailu
Code Pull Requests Activity

Initial commit: FastMCP server for Mailu email API integration

Browse Source 
- Complete FastMCP server with OpenAPI integration and fallback tools
- Automatic tool generation from Mailu REST API endpoints
- Bearer token authentication support
- Comprehensive test suite and documentation
- PyPI-ready package configuration with proper metadata
- Environment-based configuration support
- Production-ready error handling and logging
- Examples and publishing scripts included

Features:
- User management (list, create, update, delete)
- Domain management (list, create, update, delete)
- Alias management and email forwarding
- DKIM key generation
- Manager assignment for domains
- Graceful fallback when OpenAPI validation fails

Ready for Claude Desktop integration and PyPI distribution.
This commit is contained in:
Ryan Malloy 2025-07-16 11:55:44 -06:00
commit 66d1c0732a
13 changed files with 3357 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.env.example Normal file
View File

@ -0,0 +1,9 @@
# Mailu MCP Server Configuration
# Copy this file to .env and fill in your actual values
# Mailu server configuration
MAILU_BASE_URL=https://mail.example.com
MAILU_API_TOKEN=your_api_token_here
# Optional: Enable debug logging
DEBUG=false

128
.gitignore vendored Normal file
View File

@ -0,0 +1,128 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# uv
.uv/
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db

21
LICENSE Normal file
View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 MCP Mailu Server Contributors
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

8
MANIFEST.in Normal file
View File

@ -0,0 +1,8 @@
include README.md
include LICENSE
include .env.example
recursive-include examples *.py
recursive-include tests *.py
global-exclude __pycache__
global-exclude *.py[co]
global-exclude .DS_Store

283
README.md Normal file
View File

@ -0,0 +1,283 @@
# MCP Mailu Server
A FastMCP server for integrating with Mailu email server administration using the official Mailu REST API.
This server automatically converts the entire Mailu API into MCP tools and resources, providing seamless integration between AI assistants and your Mailu mail server.
## Features
### 🔧 **Automatically Generated Tools** (from Mailu API)
- **User Management**: Create, update, delete, and find users
- **Domain Management**: Create, update, delete, and find domains
- **Alias Management**: Create, update, delete, and find email aliases
- **DKIM Operations**: Generate DKIM keys for domains
- **Manager Operations**: Assign and manage domain managers
### 📊 **Automatically Generated Resources**
- **Users**: `GET /user` - List all users
- **Domains**: `GET /domain` - List all domains
- **Aliases**: `GET /alias` - List all aliases
- **Domain Users**: `GET /domain/{domain}/users` - List users for specific domain
- **User Details**: `GET /user/{email}` - Get specific user information
- **Domain Details**: `GET /domain/{domain}` - Get specific domain information
### ⚡ **Key Benefits**
- **Complete API Coverage**: Every Mailu API endpoint automatically available
- **Type Safety**: Full TypeScript-like type checking with Pydantic models
- **Smart Resource Mapping**: GET requests become Resources, modifications become Tools
- **Authentication Handled**: Bearer token authentication built-in
- **Error Handling**: Comprehensive error responses and validation
## Installation
### From PyPI (Recommended)
```bash
# Install from PyPI
pip install mcp-mailu
# Or with uv
uv add mcp-mailu
```
### From Source
This project uses [uv](https://github.com/astral-sh/uv) for dependency management.
```bash
# Clone and navigate to the project
git clone https://git.supported.systems/MCP/mcp-mailu.git
cd mcp-mailu
# Install dependencies
uv sync
# Install in development mode
uv pip install -e .
```
## Configuration
### 1. Get Your Mailu API Token
First, you need to obtain an API token from your Mailu instance:
1. Log into your Mailu admin interface
2. Go to the API section
3. Generate a new API token
4. Copy the token for configuration
### 2. Set Environment Variables
Copy the example configuration:
```bash
cp .env.example .env
```
Edit `.env` and set your values:
```bash
# Your Mailu server URL (without /api/v1)
MAILU_BASE_URL=https://mail.yourdomain.com
# Your Mailu API token
MAILU_API_TOKEN=your_actual_token_here
# Optional: Enable debug logging
DEBUG=true
```
### 3. Alternative: Environment Variables
You can also set environment variables directly:
```bash
export MAILU_BASE_URL="https://mail.yourdomain.com"
export MAILU_API_TOKEN="your_token_here"
```
## Usage
### Running the Server
```bash
# Using uv
uv run mcp-mailu
# Or directly with Python
python -m mcp_mailu.server
# With environment variables
MAILU_BASE_URL="https://mail.example.com" MAILU_API_TOKEN="token" uv run mcp-mailu
```
### Available Operations
The server automatically exposes all Mailu API operations. Here are some examples:
#### **User Operations**
- `create_user` - Create a new email user with password, quota, and settings
- `update_user` - Update user settings, password, or quota
- `delete_user` - Remove a user account
- `find_user` - Get detailed user information
- `list_users` - Get all users across all domains
#### **Domain Operations**
- `create_domain` - Add a new email domain
- `update_domain` - Modify domain settings and quotas
- `delete_domain` - Remove a domain (and all its users)
- `find_domain` - Get domain details including DNS settings
- `list_domain` - Get all domains
- `generate_dkim` - Generate new DKIM keys for a domain
#### **Alias Operations**
- `create_alias` - Create email aliases with multiple destinations
- `update_alias` - Modify alias destinations or settings
- `delete_alias` - Remove email aliases
- `find_alias` - Get alias details
- `list_alias` - Get all aliases
#### **Examples with Parameters**
**Create a new user:**
```json
{
"email": "john.doe@example.com",
"raw_password": "SecurePassword123!",
"displayed_name": "John Doe",
"quota_bytes": 2000000000,
"enabled": true,
"comment": "Sales team member"
}
```
**Create a domain:**
```json
{
"name": "newcompany.com",
"max_users": 50,
"max_aliases": 100,
"signup_enabled": false,
"comment": "New company domain"
}
```
**Create an alias:**
```json
{
"email": "sales@example.com",
"destination": ["john@example.com", "jane@example.com"],
"comment": "Sales team alias"
}
```
## Development
### Setup Development Environment
```bash
# Install with development dependencies
uv sync --dev
# Run tests
uv run pytest
# Format code
uv run black src/
uv run ruff check src/ --fix
# Type checking
uv run mypy src/
```
### Project Structure
```
src/
├── mcp_mailu/
│ ├── __init__.py # Package initialization
│ └── server.py # Main FastMCP server with OpenAPI integration
├── .env.example # Configuration template
├── scripts/ # Build and publish scripts
├── pyproject.toml # UV project configuration
├── README.md # This file
└── .gitignore # Git ignore rules
```
### Publishing to PyPI
The project includes scripts for easy PyPI publishing:
```bash
# Build the package
python scripts/publish.py --build
# Check package validity
python scripts/publish.py --check
# Upload to TestPyPI for testing
python scripts/publish.py --test
# Upload to production PyPI
python scripts/publish.py --prod
```
Before publishing:
1. Update version in `pyproject.toml` and `src/mcp_mailu/__init__.py`
2. Update author information in `pyproject.toml`
3. Set up PyPI credentials with `uv run twine configure`
## How It Works
This server uses FastMCP's `from_openapi` method to automatically convert the Mailu REST API into an MCP server:
1. **OpenAPI Conversion**: The Mailu API OpenAPI/Swagger specification is used to generate MCP tools and resources
2. **Smart Mapping**: GET requests become MCP Resources for data retrieval, while POST/PATCH/DELETE become Tools for actions
3. **Authentication**: Bearer token authentication is automatically handled for all requests
4. **Type Safety**: All request/response models are automatically validated using Pydantic
## Troubleshooting
### Common Issues
**Authentication Errors:**
- Verify your `MAILU_API_TOKEN` is correct
- Check that the token has sufficient permissions
- Ensure your Mailu instance has the API enabled
**Connection Errors:**
- Verify `MAILU_BASE_URL` is correct and accessible
- Check firewall settings and network connectivity
- Ensure HTTPS is properly configured if using SSL
**Permission Errors:**
- Ensure your API token has admin privileges
- Check that the specific domain/user operations are permitted
### Debug Mode
Enable debug logging to troubleshoot issues:
```bash
DEBUG=true uv run mcp-mailu
```
## API Reference
The server exposes the complete Mailu API. For detailed parameter information, refer to:
- [Mailu Official Documentation](https://mailu.io/2024.06/)
- [Mailu API Swagger Specification](https://mail.supported.systems/api/v1/swagger.json)
## Contributing
1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests and linting
5. Submit a pull request
## License
This project is licensed under the MIT License.

163
examples/test_server.py Normal file
View File

@ -0,0 +1,163 @@
#!/usr/bin/env python3
"""
Example script demonstrating how to test the Mailu MCP server.
This script shows how to use the FastMCP client to connect to
the Mailu MCP server and perform operations.
"""
import asyncio
import os
from fastmcp import FastMCP
from fastmcp.client import Client
async def test_mailu_mcp_server():
"""Test the Mailu MCP server functionality."""
print("🚀 Testing Mailu MCP Server")
print("=" * 50)
# Create a client to connect to our MCP server
# Note: In practice, the server would be running separately
client = Client("stdio")
try:
# Connect to the server
print("📡 Connecting to MCP server...")
await client.connect()
# List available tools
print("\n🔧 Available Tools:")
tools = await client.list_tools()
for tool in tools:
print(f" - {tool.name}: {tool.description}")
# List available resources
print("\n📊 Available Resources:")
resources = await client.list_resources()
for resource in resources:
print(f" - {resource.uri}: {resource.name}")
# Example: List all users (if configured)
print("\n👥 Testing user listing...")
try:
result = await client.call_tool("list_users", {})
print(f"✅ Users retrieved: {len(result.get('content', []))} users")
except Exception as e:
print(f"⚠️ Note: {e} (Expected if not configured)")
# Example: List all domains
print("\n🌐 Testing domain listing...")
try:
result = await client.call_tool("list_domain", {})
print(f"✅ Domains retrieved: {len(result.get('content', []))} domains")
except Exception as e:
print(f"⚠️ Note: {e} (Expected if not configured)")
except Exception as e:
print(f"❌ Error: {e}")
print("\n💡 Tips:")
print(" 1. Make sure you have set MAILU_BASE_URL and MAILU_API_TOKEN")
print(" 2. Ensure your Mailu server is accessible")
print(" 3. Verify your API token has the required permissions")
finally:
await client.close()
async def show_configuration_example():
"""Show example configuration."""
print("\n⚙️ Configuration Example")
print("=" * 50)
example_config = """
# Copy .env.example to .env and update with your values:
MAILU_BASE_URL=https://mail.yourdomain.com
MAILU_API_TOKEN=your_actual_api_token_here
# Then run the server:
uv run mcp-mailu
"""
print(example_config)
# Check current environment
base_url = os.getenv("MAILU_BASE_URL")
api_token = os.getenv("MAILU_API_TOKEN")
print(f"Current MAILU_BASE_URL: {base_url or 'Not set'}")
print(f"Current MAILU_API_TOKEN: {'Set' if api_token else 'Not set'}")
async def demonstrate_api_operations():
"""Demonstrate typical API operations."""
print("\n📋 Common API Operations")
print("=" * 50)
operations = [
{
"name": "Create User",
"tool": "create_user",
"example": {
"email": "john.doe@example.com",
"raw_password": "SecurePassword123!",
"displayed_name": "John Doe",
"quota_bytes": 1000000000, # 1GB
"enabled": True,
"comment": "New employee"
}
},
{
"name": "Create Domain",
"tool": "create_domain",
"example": {
"name": "newcompany.com",
"max_users": 50,
"max_aliases": 100,
"signup_enabled": False,
"comment": "New company domain"
}
},
{
"name": "Create Alias",
"tool": "create_alias",
"example": {
"email": "sales@example.com",
"destination": ["john@example.com", "jane@example.com"],
"comment": "Sales team alias"
}
},
{
"name": "Update User Quota",
"tool": "update_user",
"example": {
"quota_bytes": 2000000000, # 2GB
"comment": "Increased quota for power user"
}
}
]
for op in operations:
print(f"\n🔧 {op['name']}:")
print(f" Tool: {op['tool']}")
print(f" Example parameters:")
for key, value in op['example'].items():
print(f" {key}: {value}")
async def main():
"""Main example function."""
await show_configuration_example()
await demonstrate_api_operations()
# Only try to connect if we have configuration
if os.getenv("MAILU_API_TOKEN"):
await test_mailu_mcp_server()
else:
print("\n💡 Set MAILU_API_TOKEN to test live connection")
if __name__ == "__main__":
asyncio.run(main())

99
pyproject.toml Normal file
View File

@ -0,0 +1,99 @@
[project]
name = "mcp-mailu"
version = "0.1.0"
description = "FastMCP server for Mailu email server API integration"
authors = [
{name = "Ryan Malloy", email = "ryan@supported.systems"}
]
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
keywords = [
"mcp",
"fastmcp",
"mailu",
"email",
"mail-server",
"api",
"integration",
"smtp",
"imap"
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Internet :: WWW/HTTP :: HTTP Servers",
"Topic :: Communications :: Email",
"Topic :: System :: Systems Administration",
"Operating System :: OS Independent",
]
dependencies = [
"fastmcp>=0.1.0",
"httpx>=0.25.0",
"pydantic>=2.0.0",
"python-dotenv>=1.0.0",
]
[project.urls]
Homepage = "https://git.supported.systems/MCP/mcp-mailu"
Documentation = "https://git.supported.systems/MCP/mcp-mailu#readme"
Repository = "https://git.supported.systems/MCP/mcp-mailu.git"
Issues = "https://git.supported.systems/MCP/mcp-mailu/issues"
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-asyncio>=0.21.0",
"black>=23.0.0",
"ruff>=0.1.0",
"mypy>=1.0.0",
"build>=0.10.0",
"twine>=4.0.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatchling.build.targets.sdist]
include = [
"/src",
"/tests",
"/examples",
"README.md",
"LICENSE",
"MANIFEST.in",
]
[tool.uv]
dev-dependencies = [
"pytest>=7.0.0",
"pytest-asyncio>=0.21.0",
"black>=23.0.0",
"ruff>=0.1.0",
"mypy>=1.0.0",
]
[tool.black]
line-length = 88
target-version = ['py38']
[tool.ruff]
line-length = 88
target-version = "py38"
[tool.mypy]
python_version = "3.8"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
[project.scripts]
mcp-mailu = "mcp_mailu.server:main"

125
scripts/publish.py Normal file
View File

@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""
Build and publish script for mcp-mailu package.
Usage:
python scripts/publish.py --build # Build package
python scripts/publish.py --test # Upload to TestPyPI
python scripts/publish.py --prod # Upload to PyPI
python scripts/publish.py --check # Check package
"""
import argparse
import subprocess
import sys
from pathlib import Path
def run_command(cmd: str, check: bool = True) -> subprocess.CompletedProcess:
"""Run a shell command and return the result."""
print(f"🔧 Running: {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if check and result.returncode != 0:
print(f"❌ Command failed: {cmd}")
print(f"stdout: {result.stdout}")
print(f"stderr: {result.stderr}")
sys.exit(1)
return result
def build_package():
"""Build the package for distribution."""
print("📦 Building package...")
# Clean previous builds
run_command("rm -rf dist/ build/ *.egg-info")
# Build package
run_command("uv build")
print("✅ Package built successfully!")
def check_package():
"""Check the package for common issues."""
print("🔍 Checking package...")
# Check with twine
run_command("uv run twine check dist/*")
print("✅ Package check passed!")
def upload_test():
"""Upload to TestPyPI for testing."""
print("🧪 Uploading to TestPyPI...")
run_command("uv run twine upload --repository testpypi dist/*")
print("✅ Uploaded to TestPyPI!")
print("📋 Test installation:")
print(" pip install -i https://test.pypi.org/simple/ mcp-mailu")
def upload_prod():
"""Upload to production PyPI."""
print("🚀 Uploading to PyPI...")
# Double check first
response = input("⚠️ This will upload to production PyPI. Continue? (y/N): ")
if response.lower() != 'y':
print("❌ Upload cancelled.")
return
run_command("uv run twine upload dist/*")
print("✅ Uploaded to PyPI!")
print("📋 Installation:")
print(" pip install mcp-mailu")
def main():
"""Main script entry point."""
parser = argparse.ArgumentParser(description="Build and publish mcp-mailu package")
parser.add_argument("--build", action="store_true", help="Build the package")
parser.add_argument("--check", action="store_true", help="Check the package")
parser.add_argument("--test", action="store_true", help="Upload to TestPyPI")
parser.add_argument("--prod", action="store_true", help="Upload to PyPI")
args = parser.parse_args()
if not any([args.build, args.check, args.test, args.prod]):
parser.print_help()
return
# Ensure we're in the right directory
if not Path("pyproject.toml").exists():
print("❌ Run this script from the project root directory")
sys.exit(1)
try:
if args.build:
build_package()
if args.check:
check_package()
if args.test:
build_package()
check_package()
upload_test()
if args.prod:
build_package()
check_package()
upload_prod()
except KeyboardInterrupt:
print("\n❌ Operation cancelled by user")
sys.exit(1)
if __name__ == "__main__":
main()

29
src/mcp_mailu/__init__.py Normal file
View File

@ -0,0 +1,29 @@
"""MCP Mailu server package.
A FastMCP server for integrating with Mailu email server administration
using the official Mailu REST API.
Features:
- Complete Mailu API coverage through OpenAPI integration
- Automatic tool and resource generation
- Bearer token authentication
- Type-safe operations with Pydantic models
"""
__version__ = "0.1.0"
__author__ = "Ryan Malloy"
__email__ = "ryan@supported.systems"
__description__ = "FastMCP server for Mailu email server API integration"
# Make key components available at package level
from .server import create_mcp_server, create_mailu_client, main
__all__ = [
"__version__",
"__author__",
"__email__",
"__description__",
"create_mcp_server",
"create_mailu_client",
"main",
]

811
src/mcp_mailu/server.py Normal file
View File

@ -0,0 +1,811 @@
"""FastMCP server for Mailu integration using OpenAPI specification."""
import asyncio
import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, Optional
import httpx
from dotenv import load_dotenv
from fastmcp import FastMCP
# Load environment variables from .env file
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Mailu API OpenAPI specification
MAILU_OPENAPI_SPEC = {
"swagger": "2.0",
"basePath": "/api/v1",
"info": {
"title": "Mailu API",
"version": "1.0"
},
"host": "mail.example.com", # This will be configurable
"schemes": ["https"],
"produces": ["application/json"],
"consumes": ["application/json"],
"securityDefinitions": {
"Bearer": {
"type": "apiKey",
"in": "header",
"name": "Authorization"
}
},
"security": [{"Bearer": []}],
"paths": {
"/alias": {
"get": {
"responses": {
"200": {
"description": "Success",
"schema": {
"type": "array",
"items": {"$ref": "#/definitions/Alias"}
}
}
},
"summary": "List aliases",
"operationId": "list_alias",
"security": [{"Bearer": []}],
"tags": ["alias"]
},
"post": {
"responses": {
"409": {
"description": "Duplicate alias",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Create a new alias",
"operationId": "create_alias",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/Alias"}
}],
"security": [{"Bearer": []}],
"tags": ["alias"]
}
},
"/alias/{alias}": {
"parameters": [{
"name": "alias",
"in": "path",
"required": True,
"type": "string"
}],
"get": {
"responses": {
"404": {
"description": "Alias not found",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Alias"}
}
},
"summary": "Find alias",
"operationId": "find_alias",
"security": [{"Bearer": []}],
"tags": ["alias"]
},
"patch": {
"responses": {
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"404": {
"description": "Alias not found",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Update alias",
"operationId": "update_alias",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/AliasUpdate"}
}],
"security": [{"Bearer": []}],
"tags": ["alias"]
},
"delete": {
"responses": {
"404": {
"description": "Alias not found",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Delete alias",
"operationId": "delete_alias",
"security": [{"Bearer": []}],
"tags": ["alias"]
}
},
"/domain": {
"get": {
"responses": {
"200": {
"description": "Success",
"schema": {
"type": "array",
"items": {"$ref": "#/definitions/DomainGet"}
}
}
},
"summary": "List domains",
"operationId": "list_domain",
"security": [{"Bearer": []}],
"tags": ["domain"]
},
"post": {
"responses": {
"409": {
"description": "Duplicate domain/alternative name",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Create a new domain",
"operationId": "create_domain",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/Domain"}
}],
"security": [{"Bearer": []}],
"tags": ["domain"]
}
},
"/domain/{domain}": {
"parameters": [{
"name": "domain",
"in": "path",
"required": True,
"type": "string"
}],
"get": {
"responses": {
"404": {
"description": "Domain not found",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Domain"}
}
},
"summary": "Find domain by name",
"operationId": "find_domain",
"security": [{"Bearer": []}],
"tags": ["domain"]
},
"patch": {
"responses": {
"409": {
"description": "Duplicate domain/alternative name",
"schema": {"$ref": "#/definitions/Response"}
},
"404": {
"description": "Domain not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Update an existing domain",
"operationId": "update_domain",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/DomainUpdate"}
}],
"security": [{"Bearer": []}],
"tags": ["domain"]
},
"delete": {
"responses": {
"404": {
"description": "Domain not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Delete domain",
"operationId": "delete_domain",
"security": [{"Bearer": []}],
"tags": ["domain"]
}
},
"/domain/{domain}/users": {
"parameters": [{
"name": "domain",
"in": "path",
"required": True,
"type": "string"
}],
"get": {
"responses": {
"404": {
"description": "Domain not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {
"type": "array",
"items": {"$ref": "#/definitions/UserGet"}
}
}
},
"summary": "List users from domain",
"operationId": "list_user_domain",
"security": [{"Bearer": []}],
"tags": ["domain"]
}
},
"/user": {
"get": {
"responses": {
"200": {
"description": "Success",
"schema": {
"type": "array",
"items": {"$ref": "#/definitions/UserGet"}
}
}
},
"summary": "List users",
"operationId": "list_users",
"security": [{"Bearer": []}],
"tags": ["user"]
},
"post": {
"responses": {
"409": {
"description": "Duplicate user",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception"
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Create user",
"operationId": "create_user",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/UserCreate"}
}],
"security": [{"Bearer": []}],
"tags": ["user"]
}
},
"/user/{email}": {
"parameters": [{
"name": "email",
"in": "path",
"required": True,
"type": "string"
}],
"get": {
"responses": {
"404": {
"description": "User not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Find user",
"operationId": "find_user",
"security": [{"Bearer": []}],
"tags": ["user"]
},
"patch": {
"responses": {
"409": {
"description": "Duplicate user",
"schema": {"$ref": "#/definitions/Response"}
},
"404": {
"description": "User not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Update user",
"operationId": "update_user",
"parameters": [{
"name": "payload",
"required": True,
"in": "body",
"schema": {"$ref": "#/definitions/UserUpdate"}
}],
"security": [{"Bearer": []}],
"tags": ["user"]
},
"delete": {
"responses": {
"404": {
"description": "User not found",
"schema": {"$ref": "#/definitions/Response"}
},
"400": {
"description": "Input validation exception",
"schema": {"$ref": "#/definitions/Response"}
},
"200": {
"description": "Success",
"schema": {"$ref": "#/definitions/Response"}
}
},
"summary": "Delete user",
"operationId": "delete_user",
"security": [{"Bearer": []}],
"tags": ["user"]
}
}
},
"definitions": {
"UserCreate": {
"required": ["email", "raw_password"],
"properties": {
"email": {
"type": "string",
"description": "The email address of the user",
"example": "John.Doe@example.com"
},
"raw_password": {
"type": "string",
"description": "The raw (plain text) password of the user. Mailu will hash the password using BCRYPT-SHA256",
"example": "secret"
},
"comment": {
"type": "string",
"description": "A description for the user. This description is shown on the Users page",
"example": "my comment"
},
"quota_bytes": {
"type": "integer",
"description": "The maximum quota for the user's email box in bytes",
"example": "1000000000"
},
"global_admin": {
"type": "boolean",
"description": "Make the user a global administrator"
},
"enabled": {
"type": "boolean",
"description": "Enable the user. When an user is disabled, the user is unable to login to the Admin GUI or webmail or access his email via IMAP/POP3 or send mail"
},
"displayed_name": {
"type": "string",
"description": "The display name of the user within the Admin GUI",
"example": "John Doe"
},
"forward_enabled": {
"type": "boolean",
"description": "Enable auto forwarding"
},
"forward_destination": {
"type": "array",
"example": "Other@example.com",
"items": {
"type": "string",
"description": "Email address to forward emails to"
}
},
"reply_enabled": {
"type": "boolean",
"description": "Enable automatic replies. This is also known as out of office (ooo) or out of facility (oof) replies"
},
"reply_subject": {
"type": "string",
"description": "Optional subject for the automatic reply",
"example": "Out of office"
},
"reply_body": {
"type": "string",
"description": "The body of the automatic reply email",
"example": "Hello, I am out of office. I will respond when I am back."
}
},
"type": "object"
},
"UserGet": {
"properties": {
"email": {
"type": "string",
"description": "The email address of the user",
"example": "John.Doe@example.com"
},
"comment": {
"type": "string",
"description": "A description for the user",
"example": "my comment"
},
"quota_bytes": {
"type": "integer",
"description": "The maximum quota for the user's email box in bytes",
"example": "1000000000"
},
"global_admin": {
"type": "boolean",
"description": "Whether the user is a global administrator"
},
"enabled": {
"type": "boolean",
"description": "Whether the user is enabled"
},
"displayed_name": {
"type": "string",
"description": "The display name of the user",
"example": "John Doe"
}
},
"type": "object"
},
"UserUpdate": {
"properties": {
"raw_password": {
"type": "string",
"description": "The raw (plain text) password of the user",
"example": "secret"
},
"comment": {
"type": "string",
"description": "A description for the user",
"example": "my comment"
},
"quota_bytes": {
"type": "integer",
"description": "The maximum quota for the user's email box in bytes",
"example": "1000000000"
},
"enabled": {
"type": "boolean",
"description": "Enable or disable the user"
},
"displayed_name": {
"type": "string",
"description": "The display name of the user",
"example": "John Doe"
}
},
"type": "object"
},
"Domain": {
"required": ["name"],
"properties": {
"name": {
"type": "string",
"description": "FQDN (e.g. example.com)",
"example": "example.com"
},
"comment": {
"type": "string",
"description": "a comment"
},
"max_users": {
"type": "integer",
"description": "maximum number of users",
"default": -1,
"minimum": -1
},
"max_aliases": {
"type": "integer",
"description": "maximum number of aliases",
"default": -1,
"minimum": -1
},
"max_quota_bytes": {
"type": "integer",
"description": "maximum quota for mailbox",
"minimum": 0
},
"signup_enabled": {
"type": "boolean",
"description": "allow signup"
}
},
"type": "object"
},
"DomainGet": {
"allOf": [
{"$ref": "#/definitions/Domain"},
{
"properties": {
"dns_mx": {"type": "string"},
"dns_spf": {"type": "string"},
"dns_dkim": {"type": "string"},
"dns_dmarc": {"type": "string"}
}
}
]
},
"DomainUpdate": {
"properties": {
"comment": {
"type": "string",
"description": "a comment"
},
"max_users": {
"type": "integer",
"description": "maximum number of users",
"default": -1,
"minimum": -1
},
"max_aliases": {
"type": "integer",
"description": "maximum number of aliases",
"default": -1,
"minimum": -1
},
"signup_enabled": {
"type": "boolean",
"description": "allow signup"
}
},
"type": "object"
},
"Alias": {
"required": ["email"],
"properties": {
"email": {
"type": "string",
"description": "the alias email address",
"example": "user@example.com"
},
"destination": {
"type": "array",
"items": {
"type": "string",
"description": "alias email address",
"example": "user@example.com"
}
},
"comment": {
"type": "string",
"description": "a comment"
},
"wildcard": {
"type": "boolean",
"description": "enable SQL Like wildcard syntax"
}
},
"type": "object"
},
"AliasUpdate": {
"properties": {
"destination": {
"type": "array",
"items": {
"type": "string",
"description": "alias email address",
"example": "user@example.com"
}
},
"comment": {
"type": "string",
"description": "a comment"
},
"wildcard": {
"type": "boolean",
"description": "enable SQL Like wildcard syntax"
}
},
"type": "object"
},
"Response": {
"properties": {
"code": {"type": "integer"},
"message": {"type": "string"}
},
"type": "object"
}
}
}
def create_mailu_client(base_url: str, api_token: str) -> httpx.AsyncClient:
"""Create an authenticated HTTP client for Mailu API."""
headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
return httpx.AsyncClient(
base_url=base_url,
headers=headers,
timeout=30.0
)
async def create_mcp_server() -> FastMCP:
"""Create the MCP server with Mailu API integration."""
# Get configuration from environment variables
mailu_base_url = os.getenv("MAILU_BASE_URL", "https://mail.example.com")
mailu_api_token = os.getenv("MAILU_API_TOKEN", "")
if not mailu_api_token:
logger.warning("MAILU_API_TOKEN environment variable not set. Server will not work without authentication.")
# Create authenticated HTTP client
client = create_mailu_client(mailu_base_url + "/api/v1", mailu_api_token)
# Fetch the actual OpenAPI specification from Mailu
spec_url = "https://mail.supported.systems/api/v1/swagger.json"
logger.info(f"Fetching OpenAPI spec from: {spec_url}")
try:
async with httpx.AsyncClient() as fetch_client:
response = await fetch_client.get(spec_url)
response.raise_for_status()
spec = response.json()
# Update the spec with the actual base URL
if mailu_base_url:
from urllib.parse import urlparse
parsed = urlparse(mailu_base_url)
if "host" in spec:
spec["host"] = parsed.netloc
if "schemes" in spec:
spec["schemes"] = [parsed.scheme]
except Exception as e:
logger.error(f"Failed to fetch OpenAPI spec from {spec_url}: {e}")
logger.info("Falling back to basic MCP server without OpenAPI integration")
# Create a basic MCP server without OpenAPI
mcp = FastMCP("Mailu MCP Server")
@mcp.tool()
async def list_users() -> str:
"""List all users in the Mailu instance."""
async with client as mailu_client:
response = await mailu_client.get("/user")
return f"Users: {response.json()}"
@mcp.tool()
async def list_domains() -> str:
"""List all domains in the Mailu instance."""
async with client as mailu_client:
response = await mailu_client.get("/domain")
return f"Domains: {response.json()}"
logger.info("Created basic MCP server with manual tools")
return mcp
# Create MCP server from OpenAPI spec
try:
mcp = FastMCP.from_openapi(
client=client,
openapi_spec=spec,
name="Mailu MCP Server",
version="1.0.0"
)
logger.info(f"Created MCP server with OpenAPI integration for Mailu API at {mailu_base_url}")
return mcp
except Exception as openapi_error:
logger.error(f"Failed to create OpenAPI-based server: {openapi_error}")
logger.info("Falling back to basic MCP server with manual tools")
# Create a basic MCP server without OpenAPI
mcp = FastMCP("Mailu MCP Server")
@mcp.tool()
async def list_users() -> str:
"""List all users in the Mailu instance."""
try:
async with client as mailu_client:
response = await mailu_client.get("/user")
response.raise_for_status()
return f"Users: {response.json()}"
except Exception as e:
return f"Error listing users: {e}"
@mcp.tool()
async def list_domains() -> str:
"""List all domains in the Mailu instance."""
try:
async with client as mailu_client:
response = await mailu_client.get("/domain")
response.raise_for_status()
return f"Domains: {response.json()}"
except Exception as e:
return f"Error listing domains: {e}"
logger.info("Created basic MCP server with manual tools")
return mcp
async def main():
"""Main entry point for the MCP server."""
logger.info("Starting Mailu MCP Server")
try:
# Create and run the MCP server
mcp = await create_mcp_server()
await mcp.run(
transport="stdio",
capture_keyboard_interrupt=True
)
except Exception as e:
logger.error(f"Failed to start MCP server: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())

1
tests/__init__.py Normal file
View File

@ -0,0 +1 @@
# Tests package

230
tests/test_server.py Normal file
View File

@ -0,0 +1,230 @@
"""Tests for the MCP Mailu server."""
import pytest
import httpx
from unittest.mock import AsyncMock, Mock, patch
import os
from mcp_mailu.server import create_mailu_client, create_mcp_server, MAILU_OPENAPI_SPEC
class TestMailuClient:
"""Test cases for Mailu HTTP client creation."""
def test_create_mailu_client(self):
"""Test creation of authenticated HTTP client."""
base_url = "https://mail.example.com/api/v1"
api_token = "test-token"
client = create_mailu_client(base_url, api_token)
assert isinstance(client, httpx.AsyncClient)
assert client.base_url == base_url
assert client.headers["Authorization"] == "Bearer test-token"
assert client.headers["Content-Type"] == "application/json"
assert client.headers["Accept"] == "application/json"
class TestMCPServer:
"""Test cases for the MCP server initialization."""
@pytest.mark.asyncio
@patch.dict(os.environ, {
"MAILU_BASE_URL": "https://test.example.com",
"MAILU_API_TOKEN": "test-token"
})
async def test_create_mcp_server_with_env_vars(self):
"""Test MCP server creation with environment variables."""
with patch('mcp_mailu.server.FastMCP') as mock_fastmcp:
mock_server = Mock()
mock_fastmcp.from_openapi.return_value = mock_server
server = await create_mcp_server()
# Verify FastMCP.from_openapi was called
mock_fastmcp.from_openapi.assert_called_once()
# Get the call arguments
call_args = mock_fastmcp.from_openapi.call_args
# Verify client is an httpx.AsyncClient
assert isinstance(call_args.kwargs['client'], httpx.AsyncClient)
# Verify OpenAPI spec is passed
assert call_args.kwargs['openapi_spec'] is not None
# Verify route maps are configured
assert 'route_maps' in call_args.kwargs
assert len(call_args.kwargs['route_maps']) > 0
# Verify server info
assert call_args.kwargs['name'] == "Mailu MCP Server"
assert call_args.kwargs['version'] == "1.0.0"
@pytest.mark.asyncio
@patch.dict(os.environ, {}, clear=True)
async def test_create_mcp_server_without_env_vars(self):
"""Test MCP server creation with default values."""
with patch('mcp_mailu.server.FastMCP') as mock_fastmcp:
mock_server = Mock()
mock_fastmcp.from_openapi.return_value = mock_server
server = await create_mcp_server()
# Should still create server with defaults
mock_fastmcp.from_openapi.assert_called_once()
call_args = mock_fastmcp.from_openapi.call_args
client = call_args.kwargs['client']
# Should use default base URL
assert "mail.example.com" in str(client.base_url)
def test_openapi_spec_structure(self):
"""Test that the OpenAPI spec has required structure."""
spec = MAILU_OPENAPI_SPEC
# Basic OpenAPI structure
assert spec["swagger"] == "2.0"
assert "info" in spec
assert "paths" in spec
assert "definitions" in spec
# Required API info
assert spec["info"]["title"] == "Mailu API"
assert spec["basePath"] == "/api/v1"
# Security configuration
assert "securityDefinitions" in spec
assert "Bearer" in spec["securityDefinitions"]
# Verify key endpoints exist
assert "/user" in spec["paths"]
assert "/domain" in spec["paths"]
assert "/alias" in spec["paths"]
# Verify user endpoints have CRUD operations
user_path = spec["paths"]["/user"]
assert "get" in user_path # List users
assert "post" in user_path # Create user
user_detail_path = spec["paths"]["/user/{email}"]
assert "get" in user_detail_path # Get user
assert "patch" in user_detail_path # Update user
assert "delete" in user_detail_path # Delete user
# Verify definitions exist
assert "UserCreate" in spec["definitions"]
assert "UserGet" in spec["definitions"]
assert "UserUpdate" in spec["definitions"]
assert "Domain" in spec["definitions"]
assert "Alias" in spec["definitions"]
def test_user_create_schema(self):
"""Test UserCreate schema has required fields."""
spec = MAILU_OPENAPI_SPEC
user_create = spec["definitions"]["UserCreate"]
# Required fields
assert "email" in user_create["required"]
assert "raw_password" in user_create["required"]
# Properties
properties = user_create["properties"]
assert "email" in properties
assert "raw_password" in properties
assert "quota_bytes" in properties
assert "enabled" in properties
assert "displayed_name" in properties
# Field types
assert properties["email"]["type"] == "string"
assert properties["raw_password"]["type"] == "string"
assert properties["quota_bytes"]["type"] == "integer"
assert properties["enabled"]["type"] == "boolean"
def test_domain_schema(self):
"""Test Domain schema structure."""
spec = MAILU_OPENAPI_SPEC
domain = spec["definitions"]["Domain"]
# Required fields
assert "name" in domain["required"]
# Properties
properties = domain["properties"]
assert "name" in properties
assert "max_users" in properties
assert "max_aliases" in properties
assert "signup_enabled" in properties
# Field types
assert properties["name"]["type"] == "string"
assert properties["max_users"]["type"] == "integer"
assert properties["signup_enabled"]["type"] == "boolean"
class TestEnvironmentConfiguration:
"""Test environment variable handling."""
@patch.dict(os.environ, {
"MAILU_BASE_URL": "https://custom.mail.com",
"MAILU_API_TOKEN": "custom-token"
})
def test_environment_variables_loaded(self):
"""Test that environment variables are properly loaded."""
# These would be used in create_mcp_server
assert os.getenv("MAILU_BASE_URL") == "https://custom.mail.com"
assert os.getenv("MAILU_API_TOKEN") == "custom-token"
@patch.dict(os.environ, {}, clear=True)
def test_default_values_when_no_env_vars(self):
"""Test default values when environment variables are not set."""
base_url = os.getenv("MAILU_BASE_URL", "https://mail.example.com")
api_token = os.getenv("MAILU_API_TOKEN", "")
assert base_url == "https://mail.example.com"
assert api_token == ""
@pytest.mark.asyncio
class TestIntegration:
"""Integration tests for the complete server."""
@patch('mcp_mailu.server.FastMCP')
async def test_server_initialization_flow(self, mock_fastmcp):
"""Test the complete server initialization flow."""
mock_server = Mock()
mock_fastmcp.from_openapi.return_value = mock_server
with patch.dict(os.environ, {
"MAILU_BASE_URL": "https://test.mail.com",
"MAILU_API_TOKEN": "integration-token"
}):
server = await create_mcp_server()
# Verify server was created
assert server == mock_server
# Verify from_openapi was called with correct parameters
call_args = mock_fastmcp.from_openapi.call_args
# Check client configuration
client = call_args.kwargs['client']
assert isinstance(client, httpx.AsyncClient)
assert "test.mail.com" in str(client.base_url)
assert client.headers["Authorization"] == "Bearer integration-token"
# Check OpenAPI spec
spec = call_args.kwargs['openapi_spec']
assert spec['host'] == 'test.mail.com'
assert spec['schemes'] == ['https']
# Check route mappings
route_maps = call_args.kwargs['route_maps']
assert len(route_maps) > 0
# Verify some expected route patterns
patterns = [rm.pattern for rm in route_maps]
assert any("GET /user$" in pattern for pattern in patterns)
assert any("GET /domain$" in pattern for pattern in patterns)

1450
uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff