manus_webhook_create

Instructions

Implementation Reference

• manus_mcp/tools/webhooks.py:18-34 (handler)

  Handler function for manus_webhook_create tool; makes a POST /v2/webhook.create API call with the URL from WebhookCreateRequest.

  @manus_tool(
      name="manus_webhook_create",
      description=(
          "Register an HTTPS webhook URL to receive task_created and task_stopped events. "
          "Manus sends a test request before activation; endpoint must return 2xx within 10 seconds."
      ),
      input_schema=WebhookCreateRequest,
      output_schema=WebhookCreateResponse,
  )
  async def webhook_create(req: WebhookCreateRequest, ctx: ToolCtx) -> WebhookCreateResponse:
      return await ctx.client.call(
          "POST",
          "/v2/webhook.create",
          json_body=req,
          response_model=WebhookCreateResponse,
          rate_limit_key="webhook.create",
      )

• manus_mcp/schemas/webhooks.py:22-27 (schema)

  Input schema (WebhookCreateRequest with url field) and output schema (WebhookCreateResponse with webhook record) for the tool.

  class WebhookCreateRequest(ManusModel):
      url: str
  
  
  class WebhookCreateResponse(ResponseEnvelope):
      webhook: WebhookRecord

• manus_mcp/tools/registry.py:28-69 (registration)

  The @manus_tool decorator and ToolDef/ToolCtx types used by the registry to register manus_webhook_create.

  class ToolDef(Generic[TIn, TOut]):
      """Metadata + handler for a single MCP tool."""
  
      name: str
      description: str
      input_schema: type[TIn]
      output_schema: type[TOut]
      handler: Callable[[TIn, ToolCtx], Awaitable[TOut]]
      rate_limit_key: str | None = None
  
  
  _REGISTRY: dict[str, ToolDef[Any, Any]] = {}
  
  
  def manus_tool(
      *,
      name: str,
      description: str,
      input_schema: type[TIn],
      output_schema: type[TOut],
      rate_limit_key: str | None = None,
  ) -> Callable[
      [Callable[[TIn, ToolCtx], Awaitable[TOut]]], Callable[[TIn, ToolCtx], Awaitable[TOut]]
  ]:
      """Decorator registering `handler` as a tool with the given metadata."""
  
      def wrap(
          handler: Callable[[TIn, ToolCtx], Awaitable[TOut]],
      ) -> Callable[[TIn, ToolCtx], Awaitable[TOut]]:
          if name in _REGISTRY:
              raise RuntimeError(f"Duplicate tool name: {name}")
          _REGISTRY[name] = ToolDef(
              name=name,
              description=description,
              input_schema=input_schema,
              output_schema=output_schema,
              handler=handler,
              rate_limit_key=rate_limit_key,
          )
          return handler
  
      return wrap

• manus_mcp/tools/__init__.py:8-23 (registration)

  load_all_tool_modules import of webhooks module which triggers @manus_tool registration of manus_webhook_create.

  def load_all_tool_modules() -> None:
      """Import every tool module so @manus_tool decorators fire."""
      from manus_mcp.tools import (  # noqa: F401
          agents,
          browser,
          composite,
          connectors,
          files,
          projects,
          skills,
          tasks,
          usage,
          webhooks,
          website,
      )
      from manus_mcp.webhook_receiver import tools as _webhook_receiver_tools  # noqa: F401

• manus_mcp/schemas/webhooks.py:14-53 (helper)

  WebhookRecord model used in WebhookCreateResponse to represent the created webhook.

  class WebhookRecord(ManusModel):
      model_config = ConfigDict(extra="allow")
      id: str
      url: str
      status: WebhookStatus | None = None
      created_at: UnixSeconds | None = None
  
  
  class WebhookCreateRequest(ManusModel):
      url: str
  
  
  class WebhookCreateResponse(ResponseEnvelope):
      webhook: WebhookRecord
  
  
  class WebhookListQuery(ManusModel):
      pass
  
  
  class WebhookListResponse(ResponseEnvelope):
      data: list[WebhookRecord] = []
  
  
  class WebhookDeleteRequest(ManusModel):
      webhook_id: str
  
  
  class WebhookDeleteResponse(ResponseEnvelope):
      pass
  
  
  class WebhookPublicKeyQuery(ManusModel):
      pass
  
  
  class WebhookPublicKeyResponse(ResponseEnvelope):
      public_key: str
      algorithm: str | None = None