get_rewrite

Instructions

Implementation Reference

• migadu_mcp/tools/rewrite_tools.py:24-31 (handler)

  The get_rewrite tool handler. Decorated with @migadu_tool(mcp, read_only=True). Accepts 'name' (slug/name of rewrite rule) and optional 'domain', resolves domain via resolve_domain(), logs via ctx, and delegates to RewriteService.get_rewrite().

  @migadu_tool(mcp, read_only=True)
  async def get_rewrite(
      name: str, ctx: Context, domain: str | None = None
  ) -> dict[str, Any]:
      """Get rewrite rule details by slug/name."""
      resolved = resolve_domain(domain)
      await ctx.info(f"📋 Getting rewrite {name} for {resolved}")
      return await get_service_factory().rewrite_service().get_rewrite(resolved, name)

• migadu_mcp/services/rewrite_service.py:19-20 (handler)

  RewriteService.get_rewrite() - core implementation that calls the Migadu API GET /domains/{domain}/rewrites/{name} endpoint to fetch a single rewrite rule.

  async def get_rewrite(self, domain: str, name: str) -> dict[str, Any]:
      return await self.client.get(f"/domains/{domain}/rewrites/{name}")

• migadu_mcp/tools/rewrite_tools.py:16-84 (registration)

  register_rewrite_tools(mcp) - registers all rewrite tools (list_rewrites, get_rewrite, create_rewrite, update_rewrite, delete_rewrite) with the FastMCP server. get_rewrite is registered via the @migadu_tool decorator on line 24.

  def register_rewrite_tools(mcp: FastMCP) -> None:
      @migadu_tool(mcp, read_only=True, summarize_response=True)
      async def list_rewrites(ctx: Context, domain: str | None = None) -> dict[str, Any]:
          """List rewrite rules for a domain."""
          resolved = resolve_domain(domain)
          await ctx.info(f"📋 Listing rewrites for {resolved}")
          return await get_service_factory().rewrite_service().list_rewrites(resolved)
  
      @migadu_tool(mcp, read_only=True)
      async def get_rewrite(
          name: str, ctx: Context, domain: str | None = None
      ) -> dict[str, Any]:
          """Get rewrite rule details by slug/name."""
          resolved = resolve_domain(domain)
          await ctx.info(f"📋 Getting rewrite {name} for {resolved}")
          return await get_service_factory().rewrite_service().get_rewrite(resolved, name)
  
      @migadu_bulk_tool(mcp, RewriteCreateRequest, entity="rewrite", idempotent=False)
      async def create_rewrite(
          item: RewriteCreateRequest, ctx: Context
      ) -> dict[str, Any]:
          """Create rewrite rule(s). List of dicts with: name, local_part_rule (pattern), destinations, domain (optional), order_num (optional)."""
          domain = item.domain or resolve_domain(None)
          destinations = [str(d) for d in item.destinations]
          await ctx.info(f"📋 Creating rewrite {item.name} on {domain}")
          result = (
              await get_service_factory()
              .rewrite_service()
              .create_rewrite(
                  domain, item.name, item.local_part_rule, destinations, item.order_num
              )
          )
          return {"rewrite": result, "success": True}
  
      @migadu_bulk_tool(mcp, RewriteUpdateRequest, entity="rewrite")
      async def update_rewrite(
          item: RewriteUpdateRequest, ctx: Context
      ) -> dict[str, Any]:
          """Update rewrite rule(s). List of dicts with: name (required), and any of: new_name, local_part_rule, destinations, order_num, domain."""
          domain = item.domain or resolve_domain(None)
          destinations = (
              [str(d) for d in item.destinations] if item.destinations else None
          )
          await ctx.info(f"📋 Updating rewrite {item.name} on {domain}")
          result = (
              await get_service_factory()
              .rewrite_service()
              .update_rewrite(
                  domain,
                  item.name,
                  item.new_name,
                  item.local_part_rule,
                  destinations,
                  item.order_num,
              )
          )
          return {"rewrite": result, "success": True}
  
      @migadu_bulk_tool(mcp, RewriteDeleteRequest, entity="rewrite", destructive=True)
      async def delete_rewrite(
          item: RewriteDeleteRequest, ctx: Context
      ) -> dict[str, Any]:
          """Delete rewrite rule(s). DESTRUCTIVE. List of dicts with: name, domain (optional)."""
          domain = item.domain or resolve_domain(None)
          await ctx.warning(f"🗑️ Deleting rewrite {item.name}")
          await get_service_factory().rewrite_service().delete_rewrite(domain, item.name)
          return {"deleted": item.name, "success": True}
  
      _ = (list_rewrites, get_rewrite, create_rewrite, update_rewrite, delete_rewrite)

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

  Call to register_rewrite_tools(mcp) in the main server initialization sequence, which registers get_rewrite as an MCP tool.

  register_rewrite_tools(mcp)

• migadu_mcp/utils/decorators.py:34-92 (helper)

  The @migadu_tool decorator used by get_rewrite. Handles registration with FastMCP, error logging via ctx, optional response summarization, and annotation resolution.

  def migadu_tool(
      mcp: FastMCP,
      *,
      read_only: bool = False,
      destructive: bool = False,
      idempotent: bool = True,
      summarize_response: bool = False,
      max_tokens: int = DEFAULT_MAX_TOKENS,
  ) -> Callable[
      [Callable[..., Awaitable[dict[str, Any]]]], Callable[..., Awaitable[dict[str, Any]]]
  ]:
      """Register a single-target tool with FastMCP.
  
      The decorated function is called as-is with its original signature. Errors are
      logged via ctx and re-raised. If `summarize_response=True`, dict responses are
      passed through the static summarizer.
      """
      annotations = {
          "readOnlyHint": read_only,
          "destructiveHint": destructive,
          "idempotentHint": idempotent,
          "openWorldHint": True,
      }
  
      def decorator(
          func: Callable[..., Awaitable[dict[str, Any]]],
      ) -> Callable[..., Awaitable[dict[str, Any]]]:
          func_name = getattr(func, "__name__", "tool")
  
          @wraps(func)
          async def wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]:
              ctx = _extract_ctx(args, kwargs)
              try:
                  result = await func(*args, **kwargs)
              except Exception as exc:
                  if ctx is not None:
                      await ctx.error(f"❌ {func_name} failed: {exc}")
                  raise
              if summarize_response and isinstance(result, dict):
                  return summarize(result, max_tokens=max_tokens)
              return result
  
          # Preserve the original function's signature with resolved annotations so
          # FastMCP's Pydantic-based schema builder sees real types (not strings from
          # `from __future__ import annotations`). Evaluation can fail for exotic
          # annotations; in that case FastMCP falls back to the raw (string) form.
          try:
              resolved = dict(inspect.get_annotations(func, eval_str=True))
              setattr(func, "__annotations__", resolved)
          except Exception:  # nosec B110 — best-effort annotation resolution
              pass
          setattr(wrapper, "__signature__", inspect.signature(func))
          setattr(wrapper, "__annotations__", dict(getattr(func, "__annotations__", {})))
          # Register with FastMCP as a side effect; return the wrapper so the function
          # remains directly callable (tests and internal dispatch).
          mcp.tool(annotations=annotations)(wrapper)
          return wrapper
  
      return decorator