research_poll_until_finished

Instructions

Implementation Reference

• src/mcp_exa/_server.py:469-514 (handler)

  The main handler function for the 'research_poll_until_finished' tool. It is decorated with @mcp.tool(), takes parameters (research_id, poll_interval, timeout_ms, events, output_schema), builds an arguments dict, and delegates to the remote Exa MCP server via _call_mcp_tool('exa_research_poll_until_finished', arguments).

  @mcp.tool()
  def research_poll_until_finished(
      research_id: str,
      poll_interval: int | None = None,
      timeout_ms: int | None = None,
      events: bool | None = None,
      output_schema: type[BaseModel] | None = None,
  ) -> dict[str, Any]:
      """Poll until research is finished using Exa.
  
      Args:
          research_id: The unique identifier of the research task.
          poll_interval: Milliseconds between polling attempts (default: 1000).
          timeout_ms: Maximum time to wait in milliseconds (default: 600000).
          events: Whether to include events in the response.
          output_schema: Optional Pydantic model for typed output validation.
  
      Returns:
          Dict containing the completed research task.
  
      Example:
          >>> research_poll_until_finished("abc-123")
          {"research_id": "abc-123", "status": "completed", "output": {...}}
      """
      import asyncio
  
      if not research_id:
          raise ValueError("Research ID cannot be empty")
  
      arguments: dict[str, Any] = {"research_id": research_id}
      if poll_interval is not None:
          arguments["poll_interval"] = poll_interval
      if timeout_ms is not None:
          arguments["timeout_ms"] = timeout_ms
      if events is not None:
          arguments["events"] = events
      if output_schema is not None:
          arguments["output_schema"] = output_schema
  
      try:
          result = asyncio.get_event_loop().run_until_complete(
              _call_mcp_tool("exa_research_poll_until_finished", arguments)
          )
          return result
      except Exception as e:
          return {"error": str(e)}

• src/mcp_exa/_server.py:469-469 (registration)

  The tool is registered via the @mcp.tool() decorator on line 469, which registers it with the FastMCP server instance named 'mcp' (created on line 10 as fastmcp.FastMCP('mcp-exa')).

  @mcp.tool()

• src/mcp_exa/_server.py:30-69 (helper)

  The _call_mcp_tool helper function that sends JSON-RPC requests to the public Exa MCP server. It constructs a 'tools/call' request and sends it to https://mcp.exa.ai/mcp, then parses Server-Sent Events (SSE) responses.

  async def _call_mcp_tool(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
      """Call a tool on the public Exa MCP server."""
      request = {
          "jsonrpc": "2.0",
          "id": 1,
          "method": "tools/call",
          "params": {
              "name": tool_name,
              "arguments": arguments,
          },
      }
  
      async with httpx.AsyncClient(timeout=60.0) as client:
          response = await client.post(
              f"{BASE_URL}/mcp",
              json=request,
              headers={
                  "accept": "application/json, text/event-stream",
                  "content-type": "application/json",
              },
          )
          response.raise_for_status()
          response_text = response.text
  
          lines = response_text.split("\n")
          for line in lines:
              if line.startswith("data: "):
                  data = line[6:]
                  result = {"jsonrpc": "2.0", "id": 1, "result": {}}
                  try:
                      parsed = eval(data)
                  except Exception:
                      pass
                  else:
                      if "result" in parsed and parsed["result"].get("content"):
                          return {
                              "results": parsed["result"]["content"][0].get("text", "")
                          }
  
          return {"results": ""}

• src/mcp_exa/_server.py:470-476 (schema)

  Input parameter schema: research_id (str, required), poll_interval (int or None), timeout_ms (int or None), events (bool or None), output_schema (type[BaseModel] or None). Returns dict[str, Any].

  def research_poll_until_finished(
      research_id: str,
      poll_interval: int | None = None,
      timeout_ms: int | None = None,
      events: bool | None = None,
      output_schema: type[BaseModel] | None = None,
  ) -> dict[str, Any]: