get_contents

Instructions

Implementation Reference

• src/mcp_exa/_server.py:256-286 (handler)

  The 'get_contents' tool handler (registered with @mcp.tool()). Takes a URL or list of URLs, validates input, wraps in an arguments dict, and delegates to the remote Exa MCP server via _call_mcp_tool('exa_get_contents', ...).

  @mcp.tool()
  def get_contents(urls: str | list[str]) -> dict[str, Any]:
      """Retrieve contents for a list of URLs using Exa.
  
      Args:
          urls: A single URL or list of URLs to retrieve contents from.
  
      Returns:
          Dict containing the contents of each URL.
  
      Example:
          >>> get_contents(["https://example.com/article1", "https://example.com/article2"])
          {"results": [{"url": "...", "title": "...", "text": "..."}]}
      """
      import asyncio
  
      if not urls:
          raise ValueError("URLs cannot be empty")
  
      if isinstance(urls, str):
          urls = [urls]
  
      arguments: dict[str, Any] = {"urls": urls}
  
      try:
          result = asyncio.get_event_loop().run_until_complete(
              _call_mcp_tool("exa_get_contents", arguments)
          )
          return result
      except Exception as e:
          return {"error": str(e)}

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

  The tool is registered via the @mcp.tool() decorator on the get_contents function, which uses fastmcp's FastMCP instance ('mcp' created on line 10) to register the function as an MCP tool.

  @mcp.tool()

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

  The _call_mcp_tool helper function is the underlying RPC client that sends JSON-RPC requests to the public Exa MCP endpoint. get_contents calls this with tool name 'exa_get_contents' and the urls arguments.

  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": ""}