Hubble MCP Server

Instructions

Implementation Reference

• data_api.py:351-380 (handler)

  The handler function for the 'get_graph_info' tool. It is decorated with @mcp.tool() for registration and @async_retry for error handling. It sends a POST request to the Hubble API's /cluster endpoint with the provided ClusterParameters to retrieve keyword graph/cluster information.
  @mcp.tool() @async_retry(exceptions=(Exception), tries=2, delay=0.3) async def get_graph_info( req_param: ClusterParameters) -> dict[ClusterResponse, Any] | None: ''' 키워드 관계 정보(리스닝마인드의 클러스터 파인더의 결과 조회, 키워드 관계 정보 조회) 모든 키워드는 소문자로 변환하여 요청 args: req_param: ClusterParameters, 키워드 관계 정보 조회 요청 파라미터 returns: dict[ClusterResponse, Any] | None: 키워드 관계 정보 조회 결과 ClusterResponse 는 아래와 같은 정보를 포함합니다: nodes: 조회한 키워드의 앞과 뒤로 2혹은 2hop 거리 안에서 검색된 모든 키워드(노드) 리스트 nodes_count: 키워드(노드) 수 rels: 관계 리스트 rels_count: 관계 수 closeness: 관계에서 키워드가 출현한 위치. distance: 모든 관계에서 키워드가 출현한 위치. type: PEOPLE_ALSO_SEARCH_FOR | RELATED_SEARCHES | REFINEMENTS | PEOPLE_ALSO_ASK_FOR ''' async with httpx.AsyncClient() as client: headers = {"X-API-Key": HUBBLE_API_KEY} response = await client.post( f"{HUBBLE_API_URL}/cluster", headers=headers, json=req_param.model_dump(), timeout=30.0) response.raise_for_status() return response.text
• data_api.py:194-222 (schema)

  Pydantic BaseModel defining the input parameters (schema) for the get_graph_info tool, including keyword, gl, limit, hop, orientation.
  class ClusterParameters(BaseModel): keyword: str = Field( min_length=1, title="keyword(str)", description="요청 키워드", ) gl: Literal['kr', 'jp'] = Field( default='kr', title="Geolocation", description="국가 코드", ) limit: int = Field(default=1000, ge=1, le=10000, title='limit(int)', description='관계수 limit 욥션') hop: int = Field( default=2, ge=1, le=3, title="hop(int)", description="hop 수", ) orientation: Literal['UNDIRECTED', 'NATURAL', 'REVERSE'] = Field( default='UNDIRECTED', title="direction", description="관계 방향", ) _request_at: str _api_key: str def __init__(self, **data): super().__init__(**data) self._request_at = datetime.now(UTC).isoformat(timespec='milliseconds') + 'Z' # yapf:disable
• data_api.py:223-248 (schema)

  Pydantic models defining the output structure (schema) for the get_graph_info tool: ClusterRels, ClusterData, and ClusterResponse.
  class ClusterRels(BaseModel): closeness: int = Field(description=""" 해당 type 에서 키워드가 출현한 위치 """) distance: int = Field(description=""" 전체 type 에서 키워드가 출현한 위치 """) source: str target: str type: str = Field(description=""" 관계 type * PEOPLE_ALSO_SEARCH_FOR * RELATED_SEARCHES * REFINEMENTS * PEOPLE_ALSO_ASK_FOR """) class ClusterData(BaseModel): nodes: List[str] = Field(description="키워드(노드) 리스트") nodes_count: int = Field(description="키워드(노드) 수") rels: List[ClusterRels] rels_count: int = Field(description="관계 수") class ClusterResponse(BaseResponse): request_detail: ClusterParameters = Field(description="요청 받았던 파라미터") cost: int = Field(default=0) remain_credits: int = Field(default=-1) # serviceAPI data: Optional[ClusterData] = Field(default=None)
• data_api.py:19-38 (helper)

  Helper decorator 'async_retry' used on the handler for retrying API calls on exceptions.
  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