Ryan Malloy
4f17a4e3bc
MASSIVE UPGRADE: Added 27 comprehensive tools covering ALL Mailu API endpoints: 📧 USER ENDPOINTS (5 tools): - list_users, create_user, get_user, update_user, delete_user 🌐 DOMAIN ENDPOINTS (7 tools): - list_domains, create_domain, get_domain, update_domain, delete_domain - generate_dkim_keys, list_domain_users 👥 DOMAIN MANAGER ENDPOINTS (4 tools): - list_domain_managers, create_domain_manager, get_domain_manager, delete_domain_manager 📍 ALIAS ENDPOINTS (6 tools): - list_aliases, create_alias, get_alias, update_alias, delete_alias, find_aliases_by_domain 🔄 ALTERNATIVE DOMAIN ENDPOINTS (4 tools): - list_alternative_domains, create_alternative_domain, get_alternative_domain, delete_alternative_domain 🔗 RELAY ENDPOINTS (5 tools): - list_relays, create_relay, get_relay, update_relay, delete_relay ✨ FEATURES: - Complete parameter coverage for all API endpoints - Comprehensive error handling with try/catch blocks - Proper request body construction for create/update operations - All tools support the full Mailu API specification - Backward compatible with existing basic tools 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
1525 lines
64 KiB
Python
1525 lines
64 KiB
Python
"""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
|
|
)
|
|
|
|
|
|
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:
|
|
with httpx.Client() as fetch_client:
|
|
response = 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 comprehensive MCP server with manual tools
|
|
mcp = FastMCP("Mailu MCP Server")
|
|
|
|
# ===== USER ENDPOINTS =====
|
|
@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 create_user(email: str, raw_password: str, comment: str = "", quota_bytes: int = 0,
|
|
global_admin: bool = False, enabled: bool = True, enable_imap: bool = True,
|
|
enable_pop: bool = True, allow_spoofing: bool = False, forward_enabled: bool = False,
|
|
forward_destination: str = "", forward_keep: bool = True, reply_enabled: bool = False,
|
|
reply_subject: str = "", reply_body: str = "", displayed_name: str = "",
|
|
spam_enabled: bool = True, spam_mark_as_read: bool = False, spam_threshold: int = 80) -> str:
|
|
"""Create a new user in the Mailu instance."""
|
|
async with client as mailu_client:
|
|
user_data = {
|
|
"email": email,
|
|
"raw_password": raw_password,
|
|
"comment": comment,
|
|
"quota_bytes": quota_bytes,
|
|
"global_admin": global_admin,
|
|
"enabled": enabled,
|
|
"enable_imap": enable_imap,
|
|
"enable_pop": enable_pop,
|
|
"allow_spoofing": allow_spoofing,
|
|
"forward_enabled": forward_enabled,
|
|
"forward_destination": forward_destination,
|
|
"forward_keep": forward_keep,
|
|
"reply_enabled": reply_enabled,
|
|
"reply_subject": reply_subject,
|
|
"reply_body": reply_body,
|
|
"displayed_name": displayed_name,
|
|
"spam_enabled": spam_enabled,
|
|
"spam_mark_as_read": spam_mark_as_read,
|
|
"spam_threshold": spam_threshold
|
|
}
|
|
response = await mailu_client.post("/user", json=user_data)
|
|
return f"Create user result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_user(email: str) -> str:
|
|
"""Get details of a specific user."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/user/{email}")
|
|
return f"User details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def update_user(email: str, raw_password: str = "", comment: str = "", quota_bytes: int = None,
|
|
global_admin: bool = None, enabled: bool = None, enable_imap: bool = None,
|
|
enable_pop: bool = None, allow_spoofing: bool = None, forward_enabled: bool = None,
|
|
forward_destination: str = "", forward_keep: bool = None, reply_enabled: bool = None,
|
|
reply_subject: str = "", reply_body: str = "", displayed_name: str = "",
|
|
spam_enabled: bool = None, spam_mark_as_read: bool = None, spam_threshold: int = None) -> str:
|
|
"""Update an existing user."""
|
|
async with client as mailu_client:
|
|
user_data = {}
|
|
if raw_password: user_data["raw_password"] = raw_password
|
|
if comment: user_data["comment"] = comment
|
|
if quota_bytes is not None: user_data["quota_bytes"] = quota_bytes
|
|
if global_admin is not None: user_data["global_admin"] = global_admin
|
|
if enabled is not None: user_data["enabled"] = enabled
|
|
if enable_imap is not None: user_data["enable_imap"] = enable_imap
|
|
if enable_pop is not None: user_data["enable_pop"] = enable_pop
|
|
if allow_spoofing is not None: user_data["allow_spoofing"] = allow_spoofing
|
|
if forward_enabled is not None: user_data["forward_enabled"] = forward_enabled
|
|
if forward_destination: user_data["forward_destination"] = forward_destination
|
|
if forward_keep is not None: user_data["forward_keep"] = forward_keep
|
|
if reply_enabled is not None: user_data["reply_enabled"] = reply_enabled
|
|
if reply_subject: user_data["reply_subject"] = reply_subject
|
|
if reply_body: user_data["reply_body"] = reply_body
|
|
if displayed_name: user_data["displayed_name"] = displayed_name
|
|
if spam_enabled is not None: user_data["spam_enabled"] = spam_enabled
|
|
if spam_mark_as_read is not None: user_data["spam_mark_as_read"] = spam_mark_as_read
|
|
if spam_threshold is not None: user_data["spam_threshold"] = spam_threshold
|
|
|
|
response = await mailu_client.patch(f"/user/{email}", json=user_data)
|
|
return f"Update user result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_user(email: str) -> str:
|
|
"""Delete a user from the Mailu instance."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/user/{email}")
|
|
return f"Delete user result: {response.json()}"
|
|
|
|
# ===== DOMAIN ENDPOINTS =====
|
|
@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()}"
|
|
|
|
@mcp.tool()
|
|
async def create_domain(name: str, comment: str = "", max_users: int = -1, max_aliases: int = -1,
|
|
max_quota_bytes: int = 0, signup_enabled: bool = False, alternatives: str = "") -> str:
|
|
"""Create a new domain in the Mailu instance."""
|
|
async with client as mailu_client:
|
|
domain_data = {
|
|
"name": name,
|
|
"comment": comment,
|
|
"max_users": max_users,
|
|
"max_aliases": max_aliases,
|
|
"max_quota_bytes": max_quota_bytes,
|
|
"signup_enabled": signup_enabled
|
|
}
|
|
if alternatives:
|
|
domain_data["alternatives"] = alternatives.split(",")
|
|
|
|
response = await mailu_client.post("/domain", json=domain_data)
|
|
return f"Create domain result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_domain(domain: str) -> str:
|
|
"""Get details of a specific domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}")
|
|
return f"Domain details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def update_domain(domain: str, comment: str = "", max_users: int = None, max_aliases: int = None,
|
|
max_quota_bytes: int = None, signup_enabled: bool = None, alternatives: str = "") -> str:
|
|
"""Update an existing domain."""
|
|
async with client as mailu_client:
|
|
domain_data = {}
|
|
if comment: domain_data["comment"] = comment
|
|
if max_users is not None: domain_data["max_users"] = max_users
|
|
if max_aliases is not None: domain_data["max_aliases"] = max_aliases
|
|
if max_quota_bytes is not None: domain_data["max_quota_bytes"] = max_quota_bytes
|
|
if signup_enabled is not None: domain_data["signup_enabled"] = signup_enabled
|
|
if alternatives: domain_data["alternatives"] = alternatives.split(",")
|
|
|
|
response = await mailu_client.patch(f"/domain/{domain}", json=domain_data)
|
|
return f"Update domain result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_domain(domain: str) -> str:
|
|
"""Delete a domain from the Mailu instance."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/domain/{domain}")
|
|
return f"Delete domain result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def generate_dkim_keys(domain: str) -> str:
|
|
"""Generate DKIM keys for a domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.post(f"/domain/{domain}/dkim")
|
|
return f"Generate DKIM keys result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def list_domain_users(domain: str) -> str:
|
|
"""List all users in a specific domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/users")
|
|
return f"Domain users: {response.json()}"
|
|
|
|
# ===== DOMAIN MANAGER ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_domain_managers(domain: str) -> str:
|
|
"""List all managers for a specific domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/manager")
|
|
return f"Domain managers: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def create_domain_manager(domain: str, user_email: str) -> str:
|
|
"""Create a new domain manager."""
|
|
async with client as mailu_client:
|
|
manager_data = {"user_email": user_email}
|
|
response = await mailu_client.post(f"/domain/{domain}/manager", json=manager_data)
|
|
return f"Create domain manager result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_domain_manager(domain: str, email: str) -> str:
|
|
"""Get details of a specific domain manager."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/manager/{email}")
|
|
return f"Domain manager details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_domain_manager(domain: str, email: str) -> str:
|
|
"""Delete a domain manager."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/domain/{domain}/manager/{email}")
|
|
return f"Delete domain manager result: {response.json()}"
|
|
|
|
# ===== ALIAS ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_aliases() -> str:
|
|
"""List all aliases in the Mailu instance."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/alias")
|
|
return f"Aliases: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def create_alias(email: str, destination: str = "", comment: str = "", wildcard: bool = False) -> str:
|
|
"""Create a new alias."""
|
|
async with client as mailu_client:
|
|
alias_data = {
|
|
"email": email,
|
|
"destination": destination,
|
|
"comment": comment,
|
|
"wildcard": wildcard
|
|
}
|
|
response = await mailu_client.post("/alias", json=alias_data)
|
|
return f"Create alias result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_alias(alias: str) -> str:
|
|
"""Get details of a specific alias."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alias/{alias}")
|
|
return f"Alias details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def update_alias(alias: str, destination: str = "", comment: str = "", wildcard: bool = None) -> str:
|
|
"""Update an existing alias."""
|
|
async with client as mailu_client:
|
|
alias_data = {}
|
|
if destination: alias_data["destination"] = destination
|
|
if comment: alias_data["comment"] = comment
|
|
if wildcard is not None: alias_data["wildcard"] = wildcard
|
|
|
|
response = await mailu_client.patch(f"/alias/{alias}", json=alias_data)
|
|
return f"Update alias result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_alias(alias: str) -> str:
|
|
"""Delete an alias."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/alias/{alias}")
|
|
return f"Delete alias result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def find_aliases_by_domain(domain: str) -> str:
|
|
"""Find aliases by destination domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alias/destination/{domain}")
|
|
return f"Aliases for domain: {response.json()}"
|
|
|
|
# ===== ALTERNATIVE DOMAIN ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_alternative_domains() -> str:
|
|
"""List all alternative domains."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/alternative")
|
|
return f"Alternative domains: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def create_alternative_domain(name: str, domain: str) -> str:
|
|
"""Create a new alternative domain."""
|
|
async with client as mailu_client:
|
|
alt_data = {"name": name, "domain": domain}
|
|
response = await mailu_client.post("/alternative", json=alt_data)
|
|
return f"Create alternative domain result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_alternative_domain(alt: str) -> str:
|
|
"""Get details of a specific alternative domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alternative/{alt}")
|
|
return f"Alternative domain details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_alternative_domain(alt: str) -> str:
|
|
"""Delete an alternative domain."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/alternative/{alt}")
|
|
return f"Delete alternative domain result: {response.json()}"
|
|
|
|
# ===== RELAY ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_relays() -> str:
|
|
"""List all relays."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/relay")
|
|
return f"Relays: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def create_relay(name: str, smtp: str = "", comment: str = "") -> str:
|
|
"""Create a new relay."""
|
|
async with client as mailu_client:
|
|
relay_data = {"name": name, "smtp": smtp, "comment": comment}
|
|
response = await mailu_client.post("/relay", json=relay_data)
|
|
return f"Create relay result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def get_relay(name: str) -> str:
|
|
"""Get details of a specific relay."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/relay/{name}")
|
|
return f"Relay details: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def update_relay(name: str, smtp: str = "", comment: str = "") -> str:
|
|
"""Update an existing relay."""
|
|
async with client as mailu_client:
|
|
relay_data = {}
|
|
if smtp: relay_data["smtp"] = smtp
|
|
if comment: relay_data["comment"] = comment
|
|
|
|
response = await mailu_client.patch(f"/relay/{name}", json=relay_data)
|
|
return f"Update relay result: {response.json()}"
|
|
|
|
@mcp.tool()
|
|
async def delete_relay(name: str) -> str:
|
|
"""Delete a relay."""
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/relay/{name}")
|
|
return f"Delete relay result: {response.json()}"
|
|
|
|
logger.info("Created comprehensive MCP server with all 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 comprehensive MCP server with manual tools (with error handling)
|
|
mcp = FastMCP("Mailu MCP Server")
|
|
|
|
# ===== USER ENDPOINTS =====
|
|
@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 create_user(email: str, raw_password: str, comment: str = "", quota_bytes: int = 0,
|
|
global_admin: bool = False, enabled: bool = True, enable_imap: bool = True,
|
|
enable_pop: bool = True, allow_spoofing: bool = False, forward_enabled: bool = False,
|
|
forward_destination: str = "", forward_keep: bool = True, reply_enabled: bool = False,
|
|
reply_subject: str = "", reply_body: str = "", displayed_name: str = "",
|
|
spam_enabled: bool = True, spam_mark_as_read: bool = False, spam_threshold: int = 80) -> str:
|
|
"""Create a new user in the Mailu instance."""
|
|
try:
|
|
async with client as mailu_client:
|
|
user_data = {
|
|
"email": email,
|
|
"raw_password": raw_password,
|
|
"comment": comment,
|
|
"quota_bytes": quota_bytes,
|
|
"global_admin": global_admin,
|
|
"enabled": enabled,
|
|
"enable_imap": enable_imap,
|
|
"enable_pop": enable_pop,
|
|
"allow_spoofing": allow_spoofing,
|
|
"forward_enabled": forward_enabled,
|
|
"forward_destination": forward_destination,
|
|
"forward_keep": forward_keep,
|
|
"reply_enabled": reply_enabled,
|
|
"reply_subject": reply_subject,
|
|
"reply_body": reply_body,
|
|
"displayed_name": displayed_name,
|
|
"spam_enabled": spam_enabled,
|
|
"spam_mark_as_read": spam_mark_as_read,
|
|
"spam_threshold": spam_threshold
|
|
}
|
|
response = await mailu_client.post("/user", json=user_data)
|
|
response.raise_for_status()
|
|
return f"Create user result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating user: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_user(email: str) -> str:
|
|
"""Get details of a specific user."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/user/{email}")
|
|
response.raise_for_status()
|
|
return f"User details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting user: {e}"
|
|
|
|
@mcp.tool()
|
|
async def update_user(email: str, raw_password: str = "", comment: str = "", quota_bytes: int = None,
|
|
global_admin: bool = None, enabled: bool = None, enable_imap: bool = None,
|
|
enable_pop: bool = None, allow_spoofing: bool = None, forward_enabled: bool = None,
|
|
forward_destination: str = "", forward_keep: bool = None, reply_enabled: bool = None,
|
|
reply_subject: str = "", reply_body: str = "", displayed_name: str = "",
|
|
spam_enabled: bool = None, spam_mark_as_read: bool = None, spam_threshold: int = None) -> str:
|
|
"""Update an existing user."""
|
|
try:
|
|
async with client as mailu_client:
|
|
user_data = {}
|
|
if raw_password: user_data["raw_password"] = raw_password
|
|
if comment: user_data["comment"] = comment
|
|
if quota_bytes is not None: user_data["quota_bytes"] = quota_bytes
|
|
if global_admin is not None: user_data["global_admin"] = global_admin
|
|
if enabled is not None: user_data["enabled"] = enabled
|
|
if enable_imap is not None: user_data["enable_imap"] = enable_imap
|
|
if enable_pop is not None: user_data["enable_pop"] = enable_pop
|
|
if allow_spoofing is not None: user_data["allow_spoofing"] = allow_spoofing
|
|
if forward_enabled is not None: user_data["forward_enabled"] = forward_enabled
|
|
if forward_destination: user_data["forward_destination"] = forward_destination
|
|
if forward_keep is not None: user_data["forward_keep"] = forward_keep
|
|
if reply_enabled is not None: user_data["reply_enabled"] = reply_enabled
|
|
if reply_subject: user_data["reply_subject"] = reply_subject
|
|
if reply_body: user_data["reply_body"] = reply_body
|
|
if displayed_name: user_data["displayed_name"] = displayed_name
|
|
if spam_enabled is not None: user_data["spam_enabled"] = spam_enabled
|
|
if spam_mark_as_read is not None: user_data["spam_mark_as_read"] = spam_mark_as_read
|
|
if spam_threshold is not None: user_data["spam_threshold"] = spam_threshold
|
|
|
|
response = await mailu_client.patch(f"/user/{email}", json=user_data)
|
|
response.raise_for_status()
|
|
return f"Update user result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error updating user: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_user(email: str) -> str:
|
|
"""Delete a user from the Mailu instance."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/user/{email}")
|
|
response.raise_for_status()
|
|
return f"Delete user result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting user: {e}"
|
|
|
|
# ===== DOMAIN ENDPOINTS =====
|
|
@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}"
|
|
|
|
@mcp.tool()
|
|
async def create_domain(name: str, comment: str = "", max_users: int = -1, max_aliases: int = -1,
|
|
max_quota_bytes: int = 0, signup_enabled: bool = False, alternatives: str = "") -> str:
|
|
"""Create a new domain in the Mailu instance."""
|
|
try:
|
|
async with client as mailu_client:
|
|
domain_data = {
|
|
"name": name,
|
|
"comment": comment,
|
|
"max_users": max_users,
|
|
"max_aliases": max_aliases,
|
|
"max_quota_bytes": max_quota_bytes,
|
|
"signup_enabled": signup_enabled
|
|
}
|
|
if alternatives:
|
|
domain_data["alternatives"] = alternatives.split(",")
|
|
|
|
response = await mailu_client.post("/domain", json=domain_data)
|
|
response.raise_for_status()
|
|
return f"Create domain result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_domain(domain: str) -> str:
|
|
"""Get details of a specific domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}")
|
|
response.raise_for_status()
|
|
return f"Domain details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def update_domain(domain: str, comment: str = "", max_users: int = None, max_aliases: int = None,
|
|
max_quota_bytes: int = None, signup_enabled: bool = None, alternatives: str = "") -> str:
|
|
"""Update an existing domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
domain_data = {}
|
|
if comment: domain_data["comment"] = comment
|
|
if max_users is not None: domain_data["max_users"] = max_users
|
|
if max_aliases is not None: domain_data["max_aliases"] = max_aliases
|
|
if max_quota_bytes is not None: domain_data["max_quota_bytes"] = max_quota_bytes
|
|
if signup_enabled is not None: domain_data["signup_enabled"] = signup_enabled
|
|
if alternatives: domain_data["alternatives"] = alternatives.split(",")
|
|
|
|
response = await mailu_client.patch(f"/domain/{domain}", json=domain_data)
|
|
response.raise_for_status()
|
|
return f"Update domain result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error updating domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_domain(domain: str) -> str:
|
|
"""Delete a domain from the Mailu instance."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/domain/{domain}")
|
|
response.raise_for_status()
|
|
return f"Delete domain result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def generate_dkim_keys(domain: str) -> str:
|
|
"""Generate DKIM keys for a domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.post(f"/domain/{domain}/dkim")
|
|
response.raise_for_status()
|
|
return f"Generate DKIM keys result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error generating DKIM keys: {e}"
|
|
|
|
@mcp.tool()
|
|
async def list_domain_users(domain: str) -> str:
|
|
"""List all users in a specific domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/users")
|
|
response.raise_for_status()
|
|
return f"Domain users: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error listing domain users: {e}"
|
|
|
|
# ===== DOMAIN MANAGER ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_domain_managers(domain: str) -> str:
|
|
"""List all managers for a specific domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/manager")
|
|
response.raise_for_status()
|
|
return f"Domain managers: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error listing domain managers: {e}"
|
|
|
|
@mcp.tool()
|
|
async def create_domain_manager(domain: str, user_email: str) -> str:
|
|
"""Create a new domain manager."""
|
|
try:
|
|
async with client as mailu_client:
|
|
manager_data = {"user_email": user_email}
|
|
response = await mailu_client.post(f"/domain/{domain}/manager", json=manager_data)
|
|
response.raise_for_status()
|
|
return f"Create domain manager result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating domain manager: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_domain_manager(domain: str, email: str) -> str:
|
|
"""Get details of a specific domain manager."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/domain/{domain}/manager/{email}")
|
|
response.raise_for_status()
|
|
return f"Domain manager details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting domain manager: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_domain_manager(domain: str, email: str) -> str:
|
|
"""Delete a domain manager."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/domain/{domain}/manager/{email}")
|
|
response.raise_for_status()
|
|
return f"Delete domain manager result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting domain manager: {e}"
|
|
|
|
# ===== ALIAS ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_aliases() -> str:
|
|
"""List all aliases in the Mailu instance."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/alias")
|
|
response.raise_for_status()
|
|
return f"Aliases: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error listing aliases: {e}"
|
|
|
|
@mcp.tool()
|
|
async def create_alias(email: str, destination: str = "", comment: str = "", wildcard: bool = False) -> str:
|
|
"""Create a new alias."""
|
|
try:
|
|
async with client as mailu_client:
|
|
alias_data = {
|
|
"email": email,
|
|
"destination": destination,
|
|
"comment": comment,
|
|
"wildcard": wildcard
|
|
}
|
|
response = await mailu_client.post("/alias", json=alias_data)
|
|
response.raise_for_status()
|
|
return f"Create alias result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating alias: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_alias(alias: str) -> str:
|
|
"""Get details of a specific alias."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alias/{alias}")
|
|
response.raise_for_status()
|
|
return f"Alias details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting alias: {e}"
|
|
|
|
@mcp.tool()
|
|
async def update_alias(alias: str, destination: str = "", comment: str = "", wildcard: bool = None) -> str:
|
|
"""Update an existing alias."""
|
|
try:
|
|
async with client as mailu_client:
|
|
alias_data = {}
|
|
if destination: alias_data["destination"] = destination
|
|
if comment: alias_data["comment"] = comment
|
|
if wildcard is not None: alias_data["wildcard"] = wildcard
|
|
|
|
response = await mailu_client.patch(f"/alias/{alias}", json=alias_data)
|
|
response.raise_for_status()
|
|
return f"Update alias result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error updating alias: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_alias(alias: str) -> str:
|
|
"""Delete an alias."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/alias/{alias}")
|
|
response.raise_for_status()
|
|
return f"Delete alias result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting alias: {e}"
|
|
|
|
@mcp.tool()
|
|
async def find_aliases_by_domain(domain: str) -> str:
|
|
"""Find aliases by destination domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alias/destination/{domain}")
|
|
response.raise_for_status()
|
|
return f"Aliases for domain: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error finding aliases by domain: {e}"
|
|
|
|
# ===== ALTERNATIVE DOMAIN ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_alternative_domains() -> str:
|
|
"""List all alternative domains."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/alternative")
|
|
response.raise_for_status()
|
|
return f"Alternative domains: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error listing alternative domains: {e}"
|
|
|
|
@mcp.tool()
|
|
async def create_alternative_domain(name: str, domain: str) -> str:
|
|
"""Create a new alternative domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
alt_data = {"name": name, "domain": domain}
|
|
response = await mailu_client.post("/alternative", json=alt_data)
|
|
response.raise_for_status()
|
|
return f"Create alternative domain result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating alternative domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_alternative_domain(alt: str) -> str:
|
|
"""Get details of a specific alternative domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/alternative/{alt}")
|
|
response.raise_for_status()
|
|
return f"Alternative domain details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting alternative domain: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_alternative_domain(alt: str) -> str:
|
|
"""Delete an alternative domain."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/alternative/{alt}")
|
|
response.raise_for_status()
|
|
return f"Delete alternative domain result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting alternative domain: {e}"
|
|
|
|
# ===== RELAY ENDPOINTS =====
|
|
@mcp.tool()
|
|
async def list_relays() -> str:
|
|
"""List all relays."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get("/relay")
|
|
response.raise_for_status()
|
|
return f"Relays: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error listing relays: {e}"
|
|
|
|
@mcp.tool()
|
|
async def create_relay(name: str, smtp: str = "", comment: str = "") -> str:
|
|
"""Create a new relay."""
|
|
try:
|
|
async with client as mailu_client:
|
|
relay_data = {"name": name, "smtp": smtp, "comment": comment}
|
|
response = await mailu_client.post("/relay", json=relay_data)
|
|
response.raise_for_status()
|
|
return f"Create relay result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error creating relay: {e}"
|
|
|
|
@mcp.tool()
|
|
async def get_relay(name: str) -> str:
|
|
"""Get details of a specific relay."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.get(f"/relay/{name}")
|
|
response.raise_for_status()
|
|
return f"Relay details: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error getting relay: {e}"
|
|
|
|
@mcp.tool()
|
|
async def update_relay(name: str, smtp: str = "", comment: str = "") -> str:
|
|
"""Update an existing relay."""
|
|
try:
|
|
async with client as mailu_client:
|
|
relay_data = {}
|
|
if smtp: relay_data["smtp"] = smtp
|
|
if comment: relay_data["comment"] = comment
|
|
|
|
response = await mailu_client.patch(f"/relay/{name}", json=relay_data)
|
|
response.raise_for_status()
|
|
return f"Update relay result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error updating relay: {e}"
|
|
|
|
@mcp.tool()
|
|
async def delete_relay(name: str) -> str:
|
|
"""Delete a relay."""
|
|
try:
|
|
async with client as mailu_client:
|
|
response = await mailu_client.delete(f"/relay/{name}")
|
|
response.raise_for_status()
|
|
return f"Delete relay result: {response.json()}"
|
|
except Exception as e:
|
|
return f"Error deleting relay: {e}"
|
|
|
|
logger.info("Created comprehensive MCP server with all manual tools and error handling")
|
|
return mcp
|
|
|
|
|
|
def main():
|
|
"""Main entry point for the MCP server."""
|
|
logger.info("Starting Mailu MCP Server")
|
|
|
|
try:
|
|
# Create and run the MCP server
|
|
mcp = create_mcp_server()
|
|
|
|
mcp.run(
|
|
transport="stdio"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to start MCP server: {e}")
|
|
raise
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|