create_forwarding

Instructions

Implementation Reference

• migadu_mcp/tools/forwarding_tools.py:43-65 (handler)

  The MCP tool handler for 'create_forwarding' — decorated with @migadu_bulk_tool, receives a ForwardingCreateRequest, delegates to forwarding_service.create_forwarding(), and returns the result.

  @migadu_bulk_tool(
      mcp, ForwardingCreateRequest, entity="forwarding", idempotent=False
  )
  async def create_forwarding(
      item: ForwardingCreateRequest, ctx: Context
  ) -> dict[str, Any]:
      """Create forwarding(s). Forwardings require external-user confirmation. List of dicts with: mailbox, address, domain (optional), expires_on (optional, YYYY-MM-DD), remove_upon_expiry (optional)."""
      domain = item.domain or resolve_domain(None)
      await ctx.info(
          f"📋 Creating forwarding {item.address} on {item.mailbox}@{domain}"
      )
      result = (
          await get_service_factory()
          .forwarding_service()
          .create_forwarding(
              domain=domain,
              mailbox=item.mailbox,
              address=str(item.address),
              expires_on=item.expires_on.isoformat() if item.expires_on else None,
              remove_upon_expiry=item.remove_upon_expiry,
          )
      )
      return {"forwarding": result, "success": True}

• migadu_mcp/utils/schemas.py:227-241 (schema)

  The Pydantic schema ForwardingCreateRequest used for input validation of the create_forwarding tool. Includes fields: mailbox, address (EmailStr), domain, expires_on (future date validated), remove_upon_expiry.

  class ForwardingCreateRequest(BaseModel):
      mailbox: str = Field(
          ..., description="Mailbox local part the forwarding belongs to"
      )
      address: EmailStr = Field(..., description="External email address to forward to")
      domain: str | None = None
      expires_on: date | None = None
      remove_upon_expiry: bool | None = None
  
      @field_validator("expires_on")
      @classmethod
      def _future_expiry(cls, v: date | None) -> date | None:
          if v is not None and v <= date.today():
              raise ValueError("expires_on must be a future date")
          return v

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

  Registration call: register_forwarding_tools(mcp) invoked during server initialization, which sets up all forwarding tools including create_forwarding.

  register_forwarding_tools(mcp)

• migadu_mcp/services/forwarding_service.py:36-51 (helper)

  The underlying service method ForwardingService.create_forwarding() that makes the actual HTTP POST to Migadu API endpoint /domains/{domain}/mailboxes/{mailbox}/forwardings with address, expires_on, and remove_upon_expiry.

  async def create_forwarding(
      self,
      domain: str,
      mailbox: str,
      address: str,
      expires_on: str | None = None,
      remove_upon_expiry: bool | None = None,
  ) -> dict[str, Any]:
      data: dict[str, Any] = {"address": address}
      if expires_on is not None:
          data["expires_on"] = expires_on
      if remove_upon_expiry is not None:
          data["remove_upon_expiry"] = remove_upon_expiry
      return await self.client.post(
          f"/domains/{domain}/mailboxes/{mailbox}/forwardings", json=data
      )