testmo_complete_automation_run_thread
Complete an automation run thread by providing its ID. Optionally include observed and computed execution times in microseconds.
Instructions
Mark an automation run thread as completed.
Args: thread_id: The automation run thread ID to complete. elapsed_observed: Observed execution time in microseconds. elapsed_computed: Computed execution time in microseconds.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| thread_id | Yes | ||
| elapsed_observed | No | ||
| elapsed_computed | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- testmo/tools/automation.py:299-321 (handler)The tool handler function 'testmo_complete_automation_run_thread' that marks an automation run thread as completed via a POST request to the Testmo API.
@mcp.tool() async def testmo_complete_automation_run_thread( thread_id: int, elapsed_observed: int | None = None, elapsed_computed: int | None = None, ) -> dict[str, Any]: """Mark an automation run thread as completed. Args: thread_id: The automation run thread ID to complete. elapsed_observed: Observed execution time in microseconds. elapsed_computed: Computed execution time in microseconds. """ data: dict[str, Any] = {} if elapsed_observed is not None: data["elapsed_observed"] = elapsed_observed if elapsed_computed is not None: data["elapsed_computed"] = elapsed_computed return await _request( "POST", f"/automation/runs/threads/{thread_id}/complete", data=data if data else None, ) - testmo/tools/automation.py:299-300 (registration)The '@mcp.tool()' decorator registers this async function as an MCP tool named 'testmo_complete_automation_run_thread'.
@mcp.tool() async def testmo_complete_automation_run_thread( - testmo/tools/automation.py:300-304 (schema)Input parameters: thread_id (int, required), elapsed_observed (int, optional), elapsed_computed (int, optional). Output type is dict[str, Any].
async def testmo_complete_automation_run_thread( thread_id: int, elapsed_observed: int | None = None, elapsed_computed: int | None = None, ) -> dict[str, Any]: - testmo/client.py:25-49 (helper)The '_request' helper function that handles the actual HTTP call to the Testmo API.
async def _request( method: str, endpoint: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None, ) -> dict[str, Any]: async with _get_client() as client: response = await client.request( method=method, url=endpoint, json=data, params=params, ) if response.status_code == 204: return {"success": True} if response.status_code >= 400: try: error_body = response.json() except Exception: error_body = response.text raise RuntimeError( f"Testmo API error {response.status_code}: " f"{json.dumps(error_body) if isinstance(error_body, dict) else error_body}" ) return response.json() - testmo/server.py:6-6 (helper)The 'mcp' FastMCP server instance that provides the '@mcp.tool()' decorator for registration.
mcp = FastMCP("testmo-mcp")