update_domain

Instructions

Implementation Reference

• migadu_mcp/tools/domain_tools.py:58-67 (handler)

  The `update_domain` tool handler — a bulk tool decorated with @migadu_bulk_tool that receives a DomainUpdateRequest, calls domain_service().update_domain(), and returns a result envelope.

  @migadu_bulk_tool(mcp, DomainUpdateRequest, entity="domain")
  async def update_domain(item: DomainUpdateRequest, ctx: Context) -> dict[str, Any]:
      """Update domain field(s). List of dicts with: name (required), description (optional), tags (optional)."""
      await ctx.info(f"📋 Updating domain {item.name}")
      result = (
          await get_service_factory()
          .domain_service()
          .update_domain(item.name, item.description, item.tags)
      )
      return {"domain": result, "name": item.name, "success": True}

• migadu_mcp/services/domain_service.py:35-46 (helper)

  The `DomainService.update_domain()` method that builds a PATCH request body with only non-None fields (description, tags) and sends it to /domains/{name}.

  async def update_domain(
      self,
      name: str,
      description: str | None = None,
      tags: str | None = None,
  ) -> dict[str, Any]:
      data: dict[str, Any] = {}
      if description is not None:
          data["description"] = description
      if tags is not None:
          data["tags"] = tags
      return await self.client.patch(f"/domains/{name}", json=data)

• migadu_mcp/utils/schemas.py:281-284 (schema)

  The `DomainUpdateRequest` Pydantic schema defining the input: name (required), description (optional), tags (optional).

  class DomainUpdateRequest(BaseModel):
      name: str = Field(..., description="Domain name")
      description: str | None = None
      tags: str | None = None

• migadu_mcp/utils/decorators.py:95-162 (registration)

  The `@migadu_bulk_tool` decorator that wraps the handler, validates items against the schema, registers the bulk_wrapper with FastMCP, and returns bulk results.

  def migadu_bulk_tool(
      mcp: FastMCP,
      schema: Type[BaseModel],
      *,
      entity: str,
      destructive: bool = False,
      idempotent: bool = True,
  ) -> Callable[..., Any]:
      """Register a bulk mutation tool.
  
      The decorated function processes ONE validated item at a time. The outer tool
      accepts `list[dict]`, validates each, calls the inner function, and returns a
      bulk-result envelope with per-item success/failure.
      """
      annotations = {
          "readOnlyHint": False,
          "destructiveHint": destructive,
          "idempotentHint": idempotent,
          "openWorldHint": True,
      }
  
      def decorator(
          process_one: Callable[..., Awaitable[dict[str, Any]]],
      ) -> Callable[..., Awaitable[dict[str, Any]]]:
          async def bulk_wrapper(items: list, ctx: Context) -> dict:
              normalized: list[dict[str, Any]] = (
                  [items] if isinstance(items, dict) else list(items)
              )
              total = len(normalized)
              plural = entity if total == 1 else f"{entity}s"
              await ctx.info(f"🔄 Processing {total} {plural}")
  
              results: list[dict[str, Any]] = []
              for item in normalized:
                  try:
                      validated = validate_with_schema(item, schema)
                      result = await process_one(validated, ctx)
                      results.append(result)
                  except Exception as exc:
                      results.append({"error": str(exc), "item": item, "success": False})
  
              successful = sum(1 for r in results if r.get("success", True))
              failed = total - successful
  
              if failed == 0:
                  await ctx.info(f"✅ Processed {successful}/{total} {plural}")
              else:
                  await ctx.warning(
                      f"⚠️ Processed {successful}/{total} {plural}; {failed} failed"
                  )
  
              return {
                  "items": results,
                  "total_requested": total,
                  "total_successful": successful,
                  "total_failed": failed,
                  "success": failed == 0,
              }
  
          bulk_wrapper.__name__ = getattr(process_one, "__name__", "bulk_tool")
          bulk_wrapper.__qualname__ = getattr(
              process_one, "__qualname__", bulk_wrapper.__name__
          )
          bulk_wrapper.__doc__ = getattr(process_one, "__doc__", None)
          mcp.tool(annotations=annotations)(bulk_wrapper)
          return bulk_wrapper
  
      return decorator

• migadu_mcp/main.py:51-52 (registration)

  Top-level registration call: `register_domain_tools(mcp)` is invoked during server initialization.

  def initialize_server() -> None:
      register_domain_tools(mcp)