Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/identity_tools.py:100-117 (handler)

  The primary handler function for the 'create_identity' MCP tool. It handles bulk creation requests by delegating to the process_create_identity helper after logging.

      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": False,
          "openWorldHint": True,
      },
  )
  async def create_identity(
      identities: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Create email identities. List of dicts with: target, mailbox, name, password (required), domain (optional)."""
      count = len(list(ensure_iterable(identities)))
      await log_bulk_operation_start(ctx, "Creating", count, "identity")
  
      result = await process_create_identity(identities, ctx)
      await log_bulk_operation_result(ctx, "Identity creation", result, "identity")
      return result

• migadu_mcp/utils/schemas.py:189-197 (schema)

  Pydantic schema IdentityCreateRequest used by bulk_processor_with_schema for validating each identity creation input item.

  class IdentityCreateRequest(BaseModel):
      """Request schema for creating an identity"""
  
      target: str = Field(..., description="Local part of identity address")
      mailbox: str = Field(..., description="Username of mailbox that owns this identity")
      name: str = Field(..., description="Display name for identity")
      password: str = Field(..., description="Password for SMTP authentication")
      domain: Optional[str] = Field(None, description="Domain name")

• migadu_mcp/tools/identity_tools.py:67-98 (helper)

  Helper function process_create_identity that validates a single item using the schema, determines domain, formats email, calls the identity service, and logs the operation.

  @bulk_processor_with_schema(IdentityCreateRequest)
  async def process_create_identity(
      validated_item: IdentityCreateRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single identity creation"""
      # Get domain if not provided
      domain = validated_item.domain
      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, validated_item.target)
      await log_operation_start(
          ctx, "Creating identity", f"{email_address} for {validated_item.mailbox}"
      )
  
      service = get_service_factory().identity_service()
      result = await service.create_identity(
          domain,
          validated_item.mailbox,
          validated_item.target,
          validated_item.name,
          validated_item.password,
      )
  
      await log_operation_success(ctx, "Created identity", email_address)
      return {"identity": result, "email_address": email_address, "success": True}

• migadu_mcp/services/identity_service.py:22-29 (helper)

  The IdentityService.create_identity method that makes the actual API POST request to Migadu to create the identity.

  async def create_identity(
      self, domain: str, mailbox: str, local_part: str, name: str, password: str
  ) -> Dict[str, Any]:
      """Create a new identity for a mailbox"""
      data = {"local_part": local_part, "name": name, "password": password}
      return await self.client.request(
          "POST", f"/domains/{domain}/mailboxes/{mailbox}/identities", json=data
      )

• migadu_mcp/main.py:18-25 (registration)

  In main.py, register_identity_tools(mcp) is called, which defines and registers the create_identity tool via its @mcp.tool decorator.

  def initialize_server():
      """Initialize the MCP server with all tools and resources"""
      # Register all tools
      register_mailbox_tools(mcp)
      register_identity_tools(mcp)
      register_alias_tools(mcp)
      register_rewrite_tools(mcp)
      register_resources(mcp)