Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/rewrite_tools.py:130-149 (handler)

  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)

• migadu_mcp/tools/rewrite_tools.py:94-129 (helper)

  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}

• migadu_mcp/services/rewrite_service.py:20-31 (helper)

  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
      )