create_alias
Create and configure email aliases for forwarding to specified destinations using the Migadu MCP Server. Define target, domain, and internal status for efficient email management.
Instructions
Create email aliases with forwarding. List of dicts with: target (local part), destinations (email list), domain (optional), is_internal (optional).
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| aliases | Yes |
Input Schema (JSON Schema)
{
"properties": {
"aliases": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Aliases",
"type": "array"
}
},
"required": [
"aliases"
],
"type": "object"
}
Implementation Reference
- migadu_mcp/tools/alias_tools.py:145-162 (handler)MCP tool handler for create_alias: registers the tool and handles bulk alias creation by delegating to process_create_alias.@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/tools/alias_tools.py:108-144 (handler)Core handler logic for creating a single alias: validates input with AliasCreateRequest schema, calls alias service, logs operations.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/utils/schemas.py:150-164 (schema)Pydantic schema for validating create_alias input: defines fields for target, destinations, domain, is_internal with validation.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:18-25 (registration)Registration call to register_alias_tools(mcp), which registers the create_alias tool among others.def initialize_server(): """Initialize the MCP server with all tools and resources""" # Register all tools register_mailbox_tools(mcp) register_identity_tools(mcp) register_alias_tools(mcp) register_rewrite_tools(mcp) register_resources(mcp)
- Backend service method that performs the actual API call to Migadu to create the alias.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 )