Hubble MCP Server

Instructions

Implementation Reference

• data_api.py:351-380 (handler)

  The handler function for the 'get_graph_info' tool. It is registered via @mcp.tool(), uses async_retry decorator, takes ClusterParameters as input, sends a POST request to the /cluster endpoint of the Hubble API, and returns the raw response text.
  @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)

  Input schema for the get_graph_info tool, defined as Pydantic BaseModel ClusterParameters with fields for keyword, gl (geolocation), limit, hop, and 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)

  Output schema models for the get_graph_info tool: ClusterRels (relations), ClusterData (nodes and rels), and ClusterResponse (full response structure inheriting from BaseResponse).
  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)