Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/alias_tools.py:145-162 (handler)

  The main MCP tool handler function for 'create_alias'. It handles a list of alias configurations, logs the operation, delegates to process_create_alias for bulk processing, and returns the result.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": False,
          "openWorldHint": True,
      },
  )
  async def create_alias(
      aliases: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Create email aliases with forwarding. List of dicts with: target (local part), destinations (email list), domain (optional), is_internal (optional)."""
      count = len(list(ensure_iterable(aliases)))
      await log_bulk_operation_start(ctx, "Creating", count, "alias")
  
      result = await process_create_alias(aliases, ctx)
      await log_bulk_operation_result(ctx, "Alias creation", result, "alias")
      return result

• migadu_mcp/utils/schemas.py:150-164 (schema)

  Pydantic schema for validating individual alias creation requests. Used by @bulk_processor_with_schema decorator on process_create_alias.

  class AliasCreateRequest(BaseModel):
      """Request schema for creating an alias"""
  
      target: str = Field(..., description="Local part of alias")
      destinations: Union[List[EmailStr], str] = Field(
          ..., description="List of email addresses or CSV string"
      )
      domain: Optional[str] = Field(None, description="Domain name")
      is_internal: bool = Field(False, description="Internal-only flag")
  
      @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:23-23 (registration)

  Registration call to register_alias_tools(mcp), which defines and registers the create_alias tool among others.

  register_alias_tools(mcp)

• migadu_mcp/tools/alias_tools.py:107-144 (helper)

  Helper function decorated with bulk_processor_with_schema that processes a single validated alias creation request and calls the service layer.

  @bulk_processor_with_schema(AliasCreateRequest)
  async def process_create_alias(
      validated_item: AliasCreateRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single alias creation with Pydantic validation"""
      # Use validated Pydantic model directly - all validation already done
      target = validated_item.target
      destinations = validated_item.destinations
      domain = validated_item.domain
      is_internal = validated_item.is_internal
  
      # 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, "Creating 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.create_alias(
          domain, target, destinations_str, is_internal
      )
  
      await log_operation_success(ctx, "Created alias", email_address)
      if is_internal:
          await ctx.info("🔒 Configured as internal-only alias")
  
      return {"alias": result, "email_address": email_address, "success": True}

• migadu_mcp/services/alias_service.py:20-35 (helper)

  Core service implementation that performs the actual API POST request to create an alias in Migadu.

  async def create_alias(
      self,
      domain: str,
      local_part: str,
      destinations: List[str],
      is_internal: bool = False,
  ) -> Dict[str, Any]:
      """Create a new alias"""
      data = {
          "local_part": local_part,
          "destinations": ",".join(destinations),
          "is_internal": is_internal,
      }
      return await self.client.request(
          "POST", f"/domains/{domain}/aliases", json=data
      )