Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/rewrite_tools.py:187-206 (handler)

  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)

• migadu_mcp/tools/rewrite_tools.py:151-186 (helper)

  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}

• migadu_mcp/services/rewrite_service.py:37-56 (helper)

  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
      )