update_rewrite
Modify rewrite rule configurations to manage email routing. Update name, domain, new name, local part rules, and optional destinations for efficient email hosting management.
Instructions
Update rewrite rule configuration. List of dicts with: name (required), domain, new_name, local_part_rule, destinations (optional).
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| updates | Yes |
Input Schema (JSON Schema)
{
"properties": {
"updates": {
"items": {
"additionalProperties": true,
"type": "object"
},
"title": "Updates",
"type": "array"
}
},
"required": [
"updates"
],
"type": "object"
}
Implementation Reference
- The primary handler function for the MCP tool 'update_rewrite', registered via the @mcp.tool decorator. Handles bulk updates by calling the process_update_rewrite helper.@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-275 (schema)Pydantic model defining the input schema for update_rewrite operations, used by bulk_processor_with_schema for validation.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
- Helper function that processes a single validated update request, handles domain resolution, logging, and calls the rewrite service.@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}
- Core service layer implementation that makes the actual PUT request to the Migadu API 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 )