Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/alias_tools.py:145-162 (handler)

  The main MCP tool handler function for 'create_alias'. It handles a list of alias configurations, logs the operation, delegates to process_create_alias for bulk processing, and returns the result.
  @mcp.tool( annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, }, ) async def create_alias( aliases: List[Dict[str, Any]], ctx: Context ) -> Dict[str, Any]: """Create email aliases with forwarding. List of dicts with: target (local part), destinations (email list), domain (optional), is_internal (optional).""" count = len(list(ensure_iterable(aliases))) await log_bulk_operation_start(ctx, "Creating", count, "alias") result = await process_create_alias(aliases, ctx) await log_bulk_operation_result(ctx, "Alias creation", result, "alias") return result
• migadu_mcp/utils/schemas.py:150-164 (schema)

  Pydantic schema for validating individual alias creation requests. Used by @bulk_processor_with_schema decorator on process_create_alias.
  class AliasCreateRequest(BaseModel): """Request schema for creating an alias""" target: str = Field(..., description="Local part of alias") destinations: Union[List[EmailStr], str] = Field( ..., description="List of email addresses or CSV string" ) domain: Optional[str] = Field(None, description="Domain name") is_internal: bool = Field(False, description="Internal-only flag") @field_validator("destinations", mode="before") @classmethod def normalize_destinations(cls, v: Union[List[str], str]) -> List[str]: return normalize_destinations(v)
• migadu_mcp/main.py:23-23 (registration)

  Registration call to register_alias_tools(mcp), which defines and registers the create_alias tool among others.
  register_alias_tools(mcp)
• migadu_mcp/tools/alias_tools.py:107-144 (helper)

  Helper function decorated with bulk_processor_with_schema that processes a single validated alias creation request and calls the service layer.
  @bulk_processor_with_schema(AliasCreateRequest) async def process_create_alias( validated_item: AliasCreateRequest, ctx: Context ) -> Dict[str, Any]: """Process a single alias creation with Pydantic validation""" # Use validated Pydantic model directly - all validation already done target = validated_item.target destinations = validated_item.destinations domain = validated_item.domain is_internal = validated_item.is_internal # Get domain if not provided if domain is None: from migadu_mcp.config import get_config config = get_config() domain = config.get_default_domain() if not domain: raise ValueError("No domain provided and MIGADU_DOMAIN not configured") email_address = format_email_address(domain, target) await log_operation_start( ctx, "Creating alias", f"{email_address} -> {', '.join(destinations)}" ) service = get_service_factory().alias_service() # Convert List[EmailStr] to List[str] for service layer destinations_str = [str(dest) for dest in destinations] result = await service.create_alias( domain, target, destinations_str, is_internal ) await log_operation_success(ctx, "Created alias", email_address) if is_internal: await ctx.info("🔒 Configured as internal-only alias") return {"alias": result, "email_address": email_address, "success": True}
• migadu_mcp/services/alias_service.py:20-35 (helper)

  Core service implementation that performs the actual API POST request to create an alias in Migadu.
  async def create_alias( self, domain: str, local_part: str, destinations: List[str], is_internal: bool = False, ) -> Dict[str, Any]: """Create a new alias""" data = { "local_part": local_part, "destinations": ",".join(destinations), "is_internal": is_internal, } return await self.client.request( "POST", f"/domains/{domain}/aliases", json=data )