create_rewrite
Generate and configure pattern-based rewrite rules for email routing on Migadu. Define local part rules and destinations to automate email forwarding effectively.
Instructions
Create pattern-based rewrite rules. List of dicts with: name, local_part_rule, destinations (required), domain (optional).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| rewrites | Yes |
Implementation Reference
- Main MCP tool handler for 'create_rewrite'. Processes list of rewrite dicts using bulk processor, logs operations.@mcp.tool( annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, }, ) async def create_rewrite( rewrites: List[Dict[str, Any]], ctx: Context ) -> Dict[str, Any]: """Create pattern-based rewrite rules. List of dicts with: name, local_part_rule, destinations (required), domain (optional).""" count = len(list(ensure_iterable(rewrites))) await log_bulk_operation_start(ctx, "Creating", count, "rewrite rule") result = await process_create_rewrite(rewrites, ctx) await log_bulk_operation_result( ctx, "Rewrite rule creation", result, "rewrite rule" ) return result
- migadu_mcp/utils/schemas.py:233-252 (schema)Pydantic schema RewriteCreateRequest for input validation of rewrite creation requests, used by bulk_processor_with_schema.class RewriteCreateRequest(BaseModel): """Request schema for creating a rewrite rule""" name: str = Field(..., description="Unique identifier/slug for the rule") local_part_rule: str = Field( ..., description="Pattern to match (e.g., 'demo-*', 'support-*')" ) destinations: Union[List[EmailStr], str] = Field( ..., description="List of email addresses or CSV string" ) domain: Optional[str] = Field(None, description="Domain name") order_num: Optional[int] = Field( None, description="Processing order (lower numbers processed first)" ) @field_validator("destinations", mode="before") @classmethod def normalize_destinations(cls, v: Union[List[str], str]) -> List[str]: return normalize_destinations(v)
- Bulk processing helper for create_rewrite: validates input with schema, handles domain resolution, calls service layer, logs operations.@bulk_processor_with_schema(RewriteCreateRequest) async def process_create_rewrite( validated_item: RewriteCreateRequest, ctx: Context ) -> Dict[str, Any]: """Process a single rewrite rule creation with Pydantic validation""" # Use validated Pydantic model directly - all validation already done name = validated_item.name local_part_rule = validated_item.local_part_rule destinations = validated_item.destinations domain = validated_item.domain # 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") await log_operation_start( ctx, "Creating rewrite rule", f"{name}: {local_part_rule} -> {', '.join(destinations)}", ) service = get_service_factory().rewrite_service() # Convert List[EmailStr] to List[str] for service layer destinations_str = [str(dest) for dest in destinations] result = await service.create_rewrite( domain, name, local_part_rule, destinations_str ) await log_operation_success(ctx, "Created rewrite rule", f"{name}@{domain}") return {"rewrite": result, "name": name, "domain": domain, "success": True}
- Service layer implementation: constructs API payload and makes POST request to Migadu API to create rewrite.async def create_rewrite( self, domain: str, name: str, local_part_rule: str, destinations: List[str] ) -> Dict[str, Any]: """Create a new rewrite rule""" data = { "name": name, "local_part_rule": local_part_rule, "destinations": ",".join(destinations), } return await self.client.request( "POST", f"/domains/{domain}/rewrites", json=data )
- migadu_mcp/main.py:24-24 (registration)Registration of rewrite tools module (including create_rewrite handler) to the FastMCP instance in server initialization.register_rewrite_tools(mcp)