create_rewrite
Create pattern-based rewrite rules to redirect email messages based on specified criteria, enabling automated email routing and management.
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
- The main MCP tool handler for 'create_rewrite'. It processes a list of rewrite dicts using bulk processing and logging.@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 input schema for validating parameters to create_rewrite tool.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)
- migadu_mcp/main.py:24-24 (registration)Registration of rewrite tools (including create_rewrite) by calling register_rewrite_tools on the MCP instance.register_rewrite_tools(mcp)
- Helper function for processing and validating individual create_rewrite items, calls service.create_rewrite.@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}
- Low-level service implementation that makes the HTTP POST request to Migadu API to create the 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 )