crawl_google_serp

Instructions

Tool call template

Replace each placeholder with a value matching the input schema. Only required parameters are included; add optional parameters as needed.

{
  "id": 1,
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "arguments": {
      "keyword": "<keyword>"
    },
    "name": "crawl_google_serp"
  }
}

Implementation Reference

• data_api.py:381-401 (handler)

  The main handler function for the 'crawl_google_serp' MCP tool. It sends a POST request to the Hubble API's /google_serp endpoint with a keyword and geolocation parameter, returning the SERP results as text.

  @mcp.tool()
  @async_retry(exceptions=(Exception), tries=2, delay=0.3)
  async def crawl_google_serp(
          keyword: str,
          gl: Literal['kr', 'us', 'jp'] = "kr") -> dict[SerpResponse, Any] | None:
      '''
      구글 SERP API 요청
      '''
      async with httpx.AsyncClient() as client:
          payload = {
              "keyword": keyword,
              "gl": gl
          }
          headers = {"X-API-Key": HUBBLE_API_KEY}
          response = await client.post(
              f"{HUBBLE_API_URL}/google_serp",
              headers=headers,
              json=payload,
              timeout=30.0)
          response.raise_for_status()
          return response.text

• data_api.py:272-278 (schema)

  The SerpResponse Pydantic model defining the response schema for SERP API responses, including request details, cost, remaining credits, and data payload.

  class SerpResponse(BaseResponse):
      """SERP 응답 형식
      """
      request_detail: SerpParameters = Field(description="요청 받았던 파라미터")
      cost: int = Field(default=0, title="", description="")
      remain_credits: int = Field(default=-1, title="", description="")
      data: Optional[List[dict]] = Field(default=None, title="", description="")

• data_api.py:249-266 (schema)

  The SerpParameters Pydantic model defining the input/request parameters for the SERP API (keyword, geolocation, result count, device).

  class SerpParameters(BaseModel):
      """SERP 요청 형식
      """
      keyword: str = Field(min_length=1, title="요청 키워드")
      gl: Optional[Literal["kr", "jp", "us"]] = Field(
          default="kr",
          title="geo location",
      )
      num: Optional[Literal[10, 20]] = Field(
          default=20,
          title="result num",
          description="SERP 에 표출할 결과 개수",
      )
      device: Optional[Literal["mobile"]] = Field(
          default="mobile",
          title="device",
          description="(Deprecated) 요청 환경. mobile만 지원합니다.",
      )

• data_api.py:381-381 (registration)

  Registration of 'crawl_google_serp' as an MCP tool via the @mcp.tool() decorator on the FastMCP server instance 'mcp'.

  @mcp.tool()

• data_api.py:19-38 (helper)

  The async_retry decorator helper used by crawl_google_serp to retry the API call up to 2 times with a 0.3s delay on failure.

  def async_retry(exceptions=(Exception), tries=3, delay=0.3, logger=None):
      def wrapper(func):
          @wraps(func)
          async def wrapped(*args, **kwargs):
              Tries = []
              for i in range(tries):
                  try:
                      return await func(*args, **kwargs)
                  except exceptions as ex:
                      ex_msg = f"Tries({ex.__class__.__name__}) Cnt: {i+1}, {ex}"
                      Tries.append(ex_msg)
                      if logger:
                          logger.warning(ex_msg)
                      else:
                          print(ex_msg)
                      if delay:
                          await asyncio.sleep(delay)
              raise TooManyTriesException(Tries)
          return wrapped
      return wrapper