#
CapmonsterClient
The main client for interacting with the Capmonster Cloud API. Supports both synchronous and asynchronous usage, context manager protocol, and configurable polling.
#
Constructor
CapmonsterClient(api_key: str, timeout: float = 30.0, max_retries: int = 120, retry_delay: float = 2.0)
#
Context Manager
CapmonsterClient supports both sync and async context managers to ensure HTTP connections are properly closed.
with CapmonsterClient(api_key="YOUR_KEY") as client:
result = client.solve(task)
async with CapmonsterClient(api_key="YOUR_KEY") as client:
result = await client.solve_async(task)
#
Methods
#
get_balance
Fetches the current account balance.
def get_balance(self) -> float
async def get_balance_async(self) -> float
Returns: float — The account balance. Defaults to 0.0 if not found.
#
create_task
Creates a captcha-solving task and returns the task ID.
def create_task(self, task: TaskPayload, callback_url: str | None = None) -> int
async def create_task_async(self, task: TaskPayload, callback_url: str | None = None) -> int
Returns: int — The unique task identifier.
Raises: CapmonsterException if the request fails or the response contains no valid task ID.
#
get_task_result
Fetches the result of a completed task (single poll, no waiting).
def get_task_result(self, task_id: int) -> dict
async def get_task_result_async(self, task_id: int) -> dict
Returns: dict — The solution dictionary, or {} if the task is not yet ready.
#
join_task_result
Polls for the task result until it is ready, up to max_retries attempts with retry_delay between each.
def join_task_result(self, task_id: int) -> dict
async def join_task_result_async(self, task_id: int) -> dict
Returns: dict — The solution dictionary from the completed task.
Raises: CapmonsterAPIException with code ERROR_MAXIMUM_TIME_EXCEED if polling exhausts all retries.
#
solve
Convenience method that creates a task and polls for its result in one call. Equivalent to create_task() followed by join_task_result().
def solve(self, task: TaskPayload, callback_url: str | None = None) -> dict
async def solve_async(self, task: TaskPayload, callback_url: str | None = None) -> dict
Returns: dict — The solution dictionary from the completed task.
#
report_incorrect_image
Reports an incorrect image captcha solution.
def report_incorrect_image(self, task_id: int) -> None
async def report_incorrect_image_async(self, task_id: int) -> None
Raises: CapmonsterAPIException if the API returns an error.
#
report_incorrect_token
Reports an incorrect token captcha solution (reCAPTCHA, GeeTest, Turnstile, etc.).
def report_incorrect_token(self, task_id: int) -> None
async def report_incorrect_token_async(self, task_id: int) -> None
Raises: CapmonsterAPIException if the API returns an error.
#
get_user_agent
Fetches the current valid Windows User-Agent string from CapMonster Cloud.
def get_user_agent(self) -> str
async def get_user_agent_async(self) -> str
Returns: str — The current User-Agent string to use with captcha tasks.