Migadu MCP Server

Instructions

Implementation Reference

• migadu_mcp/tools/identity_tools.py:194-212 (handler)

  Main MCP tool handler for 'delete_identity', decorated with @mcp.tool(). Handles bulk deletion by calling process_delete_identity helper.
  @mcp.tool( annotations={ "readOnlyHint": False, "destructiveHint": True, "idempotentHint": True, "openWorldHint": True, }, ) async def delete_identity( targets: List[Dict[str, Any]], ctx: Context ) -> Dict[str, Any]: """Delete identities. DESTRUCTIVE: Cannot be undone. List of dicts with: target, mailbox (required), domain (optional).""" count = len(list(ensure_iterable(targets))) await log_bulk_operation_start(ctx, "Deleting", count, "identity") await ctx.warning("🗑️ DESTRUCTIVE: This operation cannot be undone!") result = await process_delete_identity(targets, ctx) await log_bulk_operation_result(ctx, "Identity deletion", result, "identity") return result
• migadu_mcp/utils/schemas.py:224-229 (schema)

  Pydantic schema for input validation of delete_identity requests, used by bulk_processor_with_schema.
  class IdentityDeleteRequest(BaseModel): """Request schema for deleting 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")
• migadu_mcp/tools/identity_tools.py:168-192 (helper)

  Helper function for processing individual validated delete requests, handles domain resolution, logging, and service call.
  @bulk_processor_with_schema(IdentityDeleteRequest) async def process_delete_identity( validated_item: IdentityDeleteRequest, ctx: Context ) -> Dict[str, Any]: """Process a single identity deletion""" # 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 ctx.warning(f"🗑️ DESTRUCTIVE: Deleting identity {email_address}") service = get_service_factory().identity_service() await service.delete_identity( domain, validated_item.mailbox, validated_item.target ) await log_operation_success(ctx, "Deleted identity", email_address) return {"deleted": email_address, "success": True}
• migadu_mcp/services/identity_service.py:63-69 (helper)

  Service method that performs the actual HTTP DELETE request to Migadu API to delete the identity.
  async def delete_identity( self, domain: str, mailbox: str, identity: str ) -> Dict[str, Any]: """Delete an identity""" return await self.client.request( "DELETE", f"/domains/{domain}/mailboxes/{mailbox}/identities/{identity}" )