Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/alias_tools.py:196-213 (handler)

  Main handler function for the 'update_alias' MCP tool, decorated with @mcp.tool. Handles bulk updates by calling process_update_alias.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  async def update_alias(
      updates: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Update alias destinations. List of dicts with: target (local part), destinations (email list), domain (optional)."""
      count = len(list(ensure_iterable(updates)))
      await log_bulk_operation_start(ctx, "Updating", count, "alias")
  
      result = await process_update_alias(updates, ctx)
      await log_bulk_operation_result(ctx, "Alias update", result, "alias")
      return result

• migadu_mcp/utils/schemas.py:166-179 (schema)

  Pydantic schema used for input validation in update_alias tool via bulk_processor_with_schema.

  class AliasUpdateRequest(BaseModel):
      """Request schema for updating an alias"""
  
      target: str = Field(..., description="Local part of alias")
      destinations: Union[List[EmailStr], str] = Field(
          ..., description="New list of email addresses or CSV string"
      )
      domain: Optional[str] = Field(None, description="Domain name")
  
      @field_validator("destinations", mode="before")
      @classmethod
      def normalize_destinations(cls, v: Union[List[str], str]) -> List[str]:
          return normalize_destinations(v)

• migadu_mcp/tools/alias_tools.py:164-195 (helper)

  Helper function that processes individual alias updates, including validation, logging, and calling the service layer.

  @bulk_processor_with_schema(AliasUpdateRequest)
  async def process_update_alias(
      validated_item: AliasUpdateRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single alias update with Pydantic validation"""
      # Use validated Pydantic model directly - all validation already done
      target = validated_item.target
      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")
  
      email_address = format_email_address(domain, target)
      await log_operation_start(
          ctx, "Updating alias", f"{email_address} -> {', '.join(destinations)}"
      )
  
      service = get_service_factory().alias_service()
      # Convert List[EmailStr] to List[str] for service layer
      destinations_str = [str(dest) for dest in destinations]
      result = await service.update_alias(domain, target, destinations_str)
  
      await log_operation_success(ctx, "Updated alias", email_address)
      return {"alias": result, "email_address": email_address, "success": True}

• migadu_mcp/services/alias_service.py:43-50 (helper)

  Service layer method that performs the actual API call to update alias destinations using MigaduClient.

  async def update_alias(
      self, domain: str, local_part: str, destinations: List[str]
  ) -> Dict[str, Any]:
      """Update alias destinations"""
      data = {"destinations": ",".join(destinations)}
      return await self.client.request(
          "PUT", f"/domains/{domain}/aliases/{local_part}", json=data
      )

• migadu_mcp/main.py:23-23 (registration)

  Registration of alias tools (including update_alias) by calling register_alias_tools on the FastMCP instance.

  register_alias_tools(mcp)