manus_website_status

Instructions

Implementation Reference

• manus_mcp/tools/website.py:16-32 (handler)

  The `website_status` async function is the actual handler for the 'manus_website_status' tool. It sends a GET request to /v2/website.status using the ManusClient with the query parameters from WebsiteStatusQuery (task_id or website_id) and returns a WebsiteStatusResponse.

  @manus_tool(
      name="manus_website_status",
      description=(
          "Get a website's publish status, live URLs, and visibility. Provide exactly one of "
          "task_id or website_id."
      ),
      input_schema=WebsiteStatusQuery,
      output_schema=WebsiteStatusResponse,
  )
  async def website_status(q: WebsiteStatusQuery, ctx: ToolCtx) -> WebsiteStatusResponse:
      return await ctx.client.call(
          "GET",
          "/v2/website.status",
          params=q.model_dump(exclude_none=True),
          response_model=WebsiteStatusResponse,
          rate_limit_key="website.status",
      )

• manus_mcp/schemas/website.py:16-30 (schema)

  WebsiteStatusQuery is the input schema (inherits from _WebsiteTarget which validates exactly one of task_id or website_id). WebsiteStatusResponse defines the output fields: website_id, publish_status, site_urls, version_id, status_updated_at, visibility.

  class _WebsiteTarget(ManusModel):
      """Helper: exactly one of task_id / website_id must be provided."""
  
      task_id: str | None = None
      website_id: str | None = None
  
      @model_validator(mode="after")
      def _one_of(self) -> _WebsiteTarget:
          if (self.task_id is None) == (self.website_id is None):
              raise ValueError("Exactly one of task_id or website_id must be provided")
          return self
  
  
  class WebsiteStatusQuery(_WebsiteTarget):
      pass

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

  The `@manus_tool` decorator registers the tool by name in the _REGISTRY dict. For 'manus_website_status', it stores a ToolDef with the handler, input_schema, and output_schema. The decorator is applied at line 16 of website.py.

  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)

  The `load_all_tool_modules` function imports the `website` module, causing the @manus_tool decorator to fire and register 'manus_website_status' (along with other website tools) into the global registry.

  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/tools/composite.py:391-472 (helper)

  The 'manus_website_publish_and_wait' composite tool internally calls the same /v2/website.status endpoint (the underlying API for 'manus_website_status') in a polling loop to wait for a deployment to complete.

  class WebsitePublishAndWaitRequest(ManusModel):
      task_id: str | None = None
      website_id: str | None = None
      visibility: WebsiteVisibility | None = None
      timeout_sec: float = Field(default=300.0, ge=1.0, le=1800.0)
      poll_interval_sec: float = Field(default=3.0, ge=0.05, le=60.0)
  
      @model_validator(mode="after")
      def _one_of(self) -> WebsitePublishAndWaitRequest:
          if (self.task_id is None) == (self.website_id is None):
              raise ValueError("Exactly one of task_id or website_id must be provided")
          return self
  
  
  class WebsitePublishAndWaitResponse(ResponseEnvelope):
      website_id: str
      deployed_version_id: str
      final_status: PublishStatus
      status: WebsiteStatusResponse
      elapsed_sec: float
  
  
  @manus_tool(
      name="manus_website_publish_and_wait",
      description=(
          "Deploy the latest checkpoint and poll until the website is published or the deployment "
          "fails. Returns the final website status with site URLs."
      ),
      input_schema=WebsitePublishAndWaitRequest,
      output_schema=WebsitePublishAndWaitResponse,
  )
  async def website_publish_and_wait(
      req: WebsitePublishAndWaitRequest, ctx: ToolCtx
  ) -> WebsitePublishAndWaitResponse:
      from manus_mcp.schemas.website import WebsitePublishResponse
  
      started = time.monotonic()
      publish_req = WebsitePublishRequest(
          task_id=req.task_id,
          website_id=req.website_id,
          visibility=req.visibility,
      )
      publish: WebsitePublishResponse = await ctx.client.call(
          "POST",
          "/v2/website.publish",
          json_body=publish_req,
          response_model=WebsitePublishResponse,
          rate_limit_key="website.publish",
      )
  
      website_id = publish.website_id
      deadline = started + req.timeout_sec
      last_status: WebsiteStatusResponse | None = None
      while True:
          status: WebsiteStatusResponse = await ctx.client.call(
              "GET",
              "/v2/website.status",
              params={"website_id": website_id},
              response_model=WebsiteStatusResponse,
              rate_limit_key="website.status",
          )
          last_status = status
          if status.publish_status in ("published", "failed"):
              break
          if time.monotonic() >= deadline:
              raise ManusApiError(
                  code="publish_timeout",
                  message=(
                      f"Website still {status.publish_status} after {req.timeout_sec}s "
                      f"(website_id={website_id})"
                  ),
              )
          await asyncio.sleep(req.poll_interval_sec)
  
      assert last_status is not None
      return WebsitePublishAndWaitResponse(
          website_id=website_id,
          deployed_version_id=publish.version_id,
          final_status=last_status.publish_status,
          status=last_status,
          elapsed_sec=round(time.monotonic() - started, 3),
      )