Bangumi TV MCP Service

Instructions

Search for persons or companies on Bangumi.

Supported Person Types (integer enum in result):
1: Individual, 2: Corporation, 3: Association

Supported Career Filters (string enum):
'producer', 'mangaka', 'artist', 'seiyu', 'writer', 'illustrator', 'actor'

Args:
    keyword: The search keyword.
    limit: Pagination limit. Defaults to 30.
    offset: Pagination offset. Defaults to 0.
    career_filter: Optional filter by person career (list of strings from PersonCareer enum).

Returns:
    Formatted search results or an error message.

Implementation Reference

• main.py:1108-1170 (handler)

  The primary handler function implementing the 'search_persons' MCP tool. It performs a POST request to the Bangumi API /v0/search/persons endpoint with keyword and optional career filters, handles errors, formats results using format_person_summary helper, and returns a formatted string summary.

  @mcp.tool()
  async def search_persons(
      keyword: str,
      limit: int = 30,
      offset: int = 0,
      career_filter: Optional[List[PersonCareer]] = None,
  ) -> str:
      """
      Search for persons or companies on Bangumi.
  
      Supported Person Types (integer enum in result):
      1: Individual, 2: Corporation, 3: Association
  
      Supported Career Filters (string enum):
      'producer', 'mangaka', 'artist', 'seiyu', 'writer', 'illustrator', 'actor'
  
      Args:
          keyword: The search keyword.
          limit: Pagination limit. Defaults to 30.
          offset: Pagination offset. Defaults to 0.
          career_filter: Optional filter by person career (list of strings from PersonCareer enum).
  
      Returns:
          Formatted search results or an error message.
      """
      json_body = {"keyword": keyword, "filter": {}}
      if career_filter:
          # Ensure string values for the API call using the enum values
          formatted_careers = [
              c.value if isinstance(c, PersonCareer) else str(c) for c in career_filter
          ]
          json_body["filter"]["career"] = formatted_careers  # Filter is in JSON body
  
      params = {"limit": limit, "offset": offset}  # Query parameters for the POST request
  
      response = await make_bangumi_request(
          method="POST",
          path="/v0/search/persons",
          query_params=params,
          json_body=json_body,
      )
  
      error_msg = handle_api_error_response(response)
      if error_msg:
          return error_msg
  
      # Expecting a dictionary with 'data' and 'total'
      if not isinstance(response, dict) or "data" not in response:
          return f"Unexpected API response format for search_persons: {response}"
  
      persons = response.get("data", [])
      if not persons:
          return f"No persons found for keyword '{keyword}'."
  
      formatted_results = [format_person_summary(p) for p in persons]
      total = response.get("total", 0)
      results_text = (
          f"Found {len(persons)} persons (Total matched: {total}).\n"
          + "---\n".join(formatted_results)
      )
  
      return results_text

• main.py:270-296 (helper)

  Helper function used by search_persons to format individual person data into a readable summary string, including type, career, summary, and image.

  def format_person_summary(person: Dict[str, Any]) -> str:
      """Formats a person dictionary into a readable summary string."""
      person_id = person.get("id")
      name = person.get("name")
      person_type = person.get("type")  # Integer enum
      career = person.get("career")  # List of string enums
      summary = person.get("short_summary") or person.get("summary", "")
  
      try:
          type_str = (
              PersonType(person_type).name if person_type is not None else "Unknown Type"
          )
      except ValueError:
          type_str = f"Unknown Type ({person_type})"
  
      formatted_string = f"[{type_str}] {name} (ID: {person_id})\n"
      if career:
          formatted_string += f"  Career: {', '.join(career)}\n"
      if summary:
          formatted_summary = summary  # [:200] + '...' if len(summary) > 200 else summary
          formatted_string += f"  Summary: {formatted_summary}\n"
  
      images = person.get("images")
      if images and images.get("common"):
          formatted_string += f"  Image: {images.get('common')}\n"
  
      return formatted_string

• main.py:77-90 (schema)

  Enum defining supported career types used as input filter for search_persons tool, providing schema/validation for career_filter parameter.

  class PersonCareer(str, Enum):
      """
      Career of a person
      'producer', 'mangaka', 'artist', 'seiyu', 'writer', 'illustrator', 'actor'
      """
  
      PRODUCER = "producer"
      MANGAKA = "mangaka"
      ARTIST = "artist"
      SEIYU = "seiyu"
      WRITER = "writer"
      ILLUSTRATOR = "illustrator"
      ACTOR = "actor"

• main.py:66-75 (schema)

  Enum defining person types returned by the API, used in formatting and validation within search_persons.

  class PersonType(IntEnum):
      """
      type of a person or company
      1 = 个人, 2 = 公司, 3 = 组合
      """
  
      INDIVIDUAL = 1
      CORPORATION = 2
      ASSOCIATION = 3

• main.py:105-169 (helper)

  Core helper function used by search_persons to make authenticated HTTP requests to the Bangumi API, with comprehensive error handling.

  async def make_bangumi_request(
      method: str,
      path: str,
      query_params: Optional[Dict[str, Any]] = None,
      json_body: Optional[Dict[str, Any]] = None,
      headers: Optional[Dict[str, str]] = None,
  ) -> Any:
      """Make a request to the Bangumi API with proper headers and error handling."""
      request_headers = headers.copy() if headers else {}
      request_headers["User-Agent"] = USER_AGENT
      request_headers["Accept"] = "application/json"
  
      if BANGUMI_TOKEN:
          request_headers["Authorization"] = f"Bearer {BANGUMI_TOKEN}"
  
      url = f"{BANGUMI_API_BASE}{path}"
  
      async with httpx.AsyncClient() as client:
          try:
              print(
                  f"DEBUG: Making {method} request to {url} with params={query_params}, json={json_body}"
              )
              response = await client.request(
                  method=method,
                  url=url,
                  params=query_params,
                  json=json_body,
                  headers=request_headers,
                  timeout=30.0,
              )
              response.raise_for_status()
              # Return the raw JSON response, let the calling tool handle its structure (dict or list)
              json_response = response.json()
              print(
                  f"DEBUG: Received response (type: {type(json_response)}, keys/length: {list(json_response.keys()) if isinstance(json_response, dict) else len(json_response) if isinstance(json_response, list) else 'N/A'})"
              )
              return json_response
          except httpx.HTTPStatusError as e:
              error_msg = (
                  f"HTTP error occurred: {e.response.status_code} - {e.response.text}"
              )
              print(f"ERROR: {error_msg}")
              # Try to parse the error response body if it's JSON
              try:
                  error_details = e.response.json()
                  return {
                      "error": error_msg,
                      "status_code": e.response.status_code,
                      "details": error_details,
                  }
              except json.JSONDecodeError:
                  return {
                      "error": error_msg,
                      "status_code": e.response.status_code,
                      "details": e.response.text,
                  }
          except httpx.RequestError as e:
              error_msg = f"An error occurred while requesting {e.request.url!r}: {e}"
              print(f"ERROR: {error_msg}")
              return {"error": error_msg}
          except Exception as e:
              error_msg = f"An unexpected error occurred: {e}"
              print(f"ERROR: {error_msg}")
              return {"error": error_msg}