Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/identity_tools.py:149-166 (handler)

  The main MCP tool handler for 'update_identity', decorated with @mcp.tool. It handles bulk updates by logging, calling the process_update_identity helper, and returning results.

  @mcp.tool(
      annotations={
          "readOnlyHint": False,
          "destructiveHint": False,
          "idempotentHint": True,
          "openWorldHint": True,
      },
  )
  async def update_identity(
      updates: List[Dict[str, Any]], ctx: Context
  ) -> Dict[str, Any]:
      """Update identity settings. List of dicts with: target, mailbox (required), domain, name, may_send, may_receive (optional)."""
      count = len(list(ensure_iterable(updates)))
      await log_bulk_operation_start(ctx, "Updating", count, "identity")
  
      result = await process_update_identity(updates, ctx)
      await log_bulk_operation_result(ctx, "Identity update", result, "identity")
      return result

• migadu_mcp/utils/schemas.py:199-222 (schema)

  Pydantic BaseModel schema 'IdentityUpdateRequest' used for input validation in the update_identity tool pipeline.

  class IdentityUpdateRequest(BaseModel):
      """Request schema for updating an identity"""
  
      target: str = Field(..., description="Local part of identity address")
      mailbox: str = Field(..., description="Username of mailbox that owns this identity")
      domain: Optional[str] = Field(None, description="Domain name")
      name: Optional[str] = Field(None, description="Update display name")
      may_send: Optional[bool] = Field(
          None, description="Allow/deny sending from identity"
      )
      may_receive: Optional[bool] = Field(
          None, description="Allow/deny receiving to identity"
      )
      may_access_imap: Optional[bool] = Field(None, description="Allow/deny IMAP access")
      may_access_pop3: Optional[bool] = Field(None, description="Allow/deny POP3 access")
      may_access_managesieve: Optional[bool] = Field(
          None, description="Allow/deny ManageSieve access"
      )
      footer_active: Optional[bool] = Field(None, description="Enable/disable footer")
      footer_plain_body: Optional[str] = Field(
          None, description="Plain text footer content"
      )
      footer_html_body: Optional[str] = Field(None, description="HTML footer content")

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

  The call to register_identity_tools(mcp) in the main server initialization, which defines and registers the update_identity tool among others.

  register_identity_tools(mcp)

• migadu_mcp/tools/identity_tools.py:118-148 (helper)

  Helper function 'process_update_identity' that handles single update: resolves domain, formats email, calls IdentityService.update_identity, and logs the operation.

  @bulk_processor_with_schema(IdentityUpdateRequest)
  async def process_update_identity(
      validated_item: IdentityUpdateRequest, ctx: Context
  ) -> Dict[str, Any]:
      """Process a single identity update"""
      # 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, "Updating identity", email_address)
  
      service = get_service_factory().identity_service()
      result = await service.update_identity(
          domain,
          validated_item.mailbox,
          validated_item.target,
          validated_item.name,
          validated_item.may_send,
          validated_item.may_receive,
      )
  
      await log_operation_success(ctx, "Updated identity", email_address)
      return {"identity": result, "email_address": email_address, "success": True}

• migadu_mcp/services/identity_service.py:39-62 (helper)

  IdentityService.update_identity method that constructs the API payload and makes the PUT request to the Migadu API to perform the actual identity update.

  async def update_identity(
      self,
      domain: str,
      mailbox: str,
      identity: str,
      name: Optional[str] = None,
      may_send: Optional[bool] = None,
      may_receive: Optional[bool] = None,
  ) -> Dict[str, Any]:
      """Update identity settings"""
      data: Dict[str, Any] = {}
      if name is not None:
          data["name"] = name
      if may_send is not None:
          data["may_send"] = may_send
      if may_receive is not None:
          data["may_receive"] = may_receive
  
      return await self.client.request(
          "PUT",
          f"/domains/{domain}/mailboxes/{mailbox}/identities/{identity}",
          json=data,
      )