update_rewrite
Modify email rewrite rules in Migadu to redirect messages based on domain, local part, or destination criteria.
Instructions
Update rewrite rule configuration. List of dicts with: name (required), domain, new_name, local_part_rule, destinations (optional).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| updates | Yes |
Implementation Reference
- MCP tool handler for the 'update_rewrite' tool. Decorated with @mcp.tool(), processes bulk updates via the helper function.@mcp.tool( annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, ) async def update_rewrite( updates: List[Dict[str, Any]], ctx: Context ) -> Dict[str, Any]: """Update rewrite rule configuration. List of dicts with: name (required), domain, new_name, local_part_rule, destinations (optional).""" count = len(list(ensure_iterable(updates))) await log_bulk_operation_start(ctx, "Updating", count, "rewrite rule") result = await process_update_rewrite(updates, ctx) await log_bulk_operation_result( ctx, "Rewrite rule update", result, "rewrite rule" ) return result
- migadu_mcp/utils/schemas.py:254-274 (schema)Pydantic schema RewriteUpdateRequest used for input validation in the bulk processor.class RewriteUpdateRequest(BaseModel): """Request schema for updating a rewrite rule""" name: str = Field(..., description="Current identifier/slug of the rule") domain: Optional[str] = Field(None, description="Domain name") new_name: Optional[str] = Field(None, description="New identifier/slug") local_part_rule: Optional[str] = Field(None, description="New pattern to match") destinations: Optional[Union[List[EmailStr], str]] = Field( None, description="New list of destinations or CSV string" ) order_num: Optional[int] = Field(None, description="New processing order") @field_validator("destinations", mode="before") @classmethod def normalize_destinations( cls, v: Optional[Union[List[str], str]] ) -> Optional[List[str]]: if v is not None: return normalize_destinations(v) return v
- migadu_mcp/main.py:23-25 (registration)Registration of rewrite tools (including update_rewrite) by calling register_rewrite_tools(mcp).register_alias_tools(mcp) register_rewrite_tools(mcp) register_resources(mcp)
- Bulk processing helper that validates input with schema and calls the service layer.@bulk_processor_with_schema(RewriteUpdateRequest) async def process_update_rewrite( validated_item: RewriteUpdateRequest, ctx: Context ) -> Dict[str, Any]: """Process a single rewrite rule update with Pydantic validation""" # Use validated Pydantic model directly - all validation already done name = validated_item.name domain = validated_item.domain new_name = validated_item.new_name local_part_rule = validated_item.local_part_rule destinations = validated_item.destinations # 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, "Updating rewrite rule", f"{name}@{domain}") service = get_service_factory().rewrite_service() # Convert Optional[List[EmailStr]] to Optional[List[str]] for service layer destinations_str = None if destinations is not None: destinations_str = [str(dest) for dest in destinations] result = await service.update_rewrite( domain, name, new_name, local_part_rule, destinations_str ) await log_operation_success(ctx, "Updated rewrite rule", f"{name}@{domain}") return {"rewrite": result, "name": name, "domain": domain, "success": True}
- Service layer implementation that makes the actual API PUT request to update the rewrite rule.async def update_rewrite( self, domain: str, name: str, new_name: Optional[str] = None, local_part_rule: Optional[str] = None, destinations: Optional[List[str]] = None, ) -> Dict[str, Any]: """Update rewrite settings""" data: Dict[str, Any] = {} if new_name is not None: data["name"] = new_name if local_part_rule is not None: data["local_part_rule"] = local_part_rule if destinations is not None: data["destinations"] = ",".join(destinations) return await self.client.request( "PUT", f"/domains/{domain}/rewrites/{name}", json=data )