create_identity

Instructions

Implementation Reference

• migadu_mcp/tools/identity_tools.py:44-59 (handler)

  The MCP tool handler for 'create_identity'. Registered via @migadu_bulk_tool decorator, it validates input against IdentityCreateRequest, resolves the domain, formats the email address, delegates to the IdentityService, and returns the result with success flag.

  @migadu_bulk_tool(mcp, IdentityCreateRequest, entity="identity", idempotent=False)
  async def create_identity(
      item: IdentityCreateRequest, ctx: Context
  ) -> dict[str, Any]:
      """Create identit(ies). List of dicts with: target, mailbox, name, password, domain (optional)."""
      domain = item.domain or resolve_domain(None)
      email = format_email_address(domain, item.target)
      await ctx.info(f"📋 Creating identity {email} on {item.mailbox}")
      result = (
          await get_service_factory()
          .identity_service()
          .create_identity(
              domain, item.mailbox, item.target, item.name, item.password
          )
      )
      return {"identity": result, "email_address": email, "success": True}

• migadu_mcp/utils/schemas.py:155-160 (schema)

  Pydantic model defining the input schema for creating an identity. Fields: target (local part), mailbox, name, password, and optional domain.

  class IdentityCreateRequest(BaseModel):
      target: str = Field(..., description="Local part of identity address")
      mailbox: str = Field(..., description="Username of mailbox that owns this identity")
      name: str
      password: str
      domain: str | None = None

• migadu_mcp/tools/identity_tools.py:17-58 (registration)

  The create_identity tool is registered via register_identity_tools(), which is called from initialize_server() in main.py. It uses @migadu_bulk_tool decorator which internally calls mcp.tool() with annotations including readOnlyHint=False and idempotentHint=False.

  def register_identity_tools(mcp: FastMCP) -> None:
      @migadu_tool(mcp, read_only=True, summarize_response=True)
      async def list_identities(
          mailbox: str, ctx: Context, domain: str | None = None
      ) -> dict[str, Any]:
          """List identities (send-as addresses) for a mailbox."""
          resolved = resolve_domain(domain)
          await ctx.info(f"📋 Listing identities for {mailbox}@{resolved}")
          return (
              await get_service_factory()
              .identity_service()
              .list_identities(resolved, mailbox)
          )
  
      @migadu_tool(mcp, read_only=True)
      async def get_identity(
          mailbox: str, identity: str, ctx: Context, domain: str | None = None
      ) -> dict[str, Any]:
          """Get full details for a specific identity."""
          resolved = resolve_domain(domain)
          await ctx.info(f"📋 Getting identity {identity}@{resolved} for {mailbox}")
          return (
              await get_service_factory()
              .identity_service()
              .get_identity(resolved, mailbox, identity)
          )
  
      @migadu_bulk_tool(mcp, IdentityCreateRequest, entity="identity", idempotent=False)
      async def create_identity(
          item: IdentityCreateRequest, ctx: Context
      ) -> dict[str, Any]:
          """Create identit(ies). List of dicts with: target, mailbox, name, password, domain (optional)."""
          domain = item.domain or resolve_domain(None)
          email = format_email_address(domain, item.target)
          await ctx.info(f"📋 Creating identity {email} on {item.mailbox}")
          result = (
              await get_service_factory()
              .identity_service()
              .create_identity(
                  domain, item.mailbox, item.target, item.name, item.password
              )
          )

• migadu_mcp/services/identity_service.py:28-39 (handler)

  The IdentityService.create_identity method that makes the actual HTTP POST request to the Migadu API endpoint /domains/{domain}/mailboxes/{mailbox}/identities with local_part, name, and password as JSON body.

  async def create_identity(
      self,
      domain: str,
      mailbox: str,
      local_part: str,
      name: str,
      password: str,
  ) -> dict[str, Any]:
      data = {"local_part": local_part, "name": name, "password": password}
      return await self.client.post(
          f"/domains/{domain}/mailboxes/{mailbox}/identities", json=data
      )