Bangumi TV MCP Service

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• main.py:755-828 (handler)

  The handler function implementing the 'get_episodes' MCP tool. It is decorated with @mcp.tool() for automatic registration. Fetches episodes from Bangumi API endpoint /v0/episodes for a given subject_id, applies optional episode_type filter, handles pagination, processes the response, formats episode details using EpType enum, and returns a formatted string summary.
  @mcp.tool() async def get_episodes( subject_id: int, episode_type: Optional[EpType] = None, limit: int = 100, offset: int = 0, ) -> str: """ List episodes for a subject. Supported Episode Types (integer enum): 0: MainStory, 1: SP, 2: OP, 3: ED, 4: PV, 5: MAD, 6: Other Args: subject_id: The ID of the subject. episode_type: Optional filter by episode type (integer value from EpType enum). limit: Pagination limit. Max 200. Defaults to 100. offset: Pagination offset. Defaults to 0. Returns: Formatted list of episodes or an error message. """ query_params: Dict[str, Any] = { "subject_id": subject_id, "limit": min(limit, 200), "offset": offset, } if episode_type is not None: query_params["type"] = int(episode_type) response = await make_bangumi_request( method="GET", path="/v0/episodes", query_params=query_params ) 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 get_episodes: {response}" episodes = response.get("data", []) if not episodes: return f"No episodes found for subject ID {subject_id} with the given criteria." formatted_results = [] for ep in episodes: ep_id = ep.get("id") name = ep.get("name") name_cn = ep.get("name_cn") sort = ep.get("sort") ep_type_int = ep.get("type") ep_type_str = "Unknown Type" if ep_type_int is not None: try: ep_type_str = EpType(ep_type_int).name except ValueError: ep_type_str = f"Unknown Type ({ep_type_int})" airdate = ep.get("airdate") formatted_results.append( f"Episode ID: {ep_id}, Type: {ep_type_str}, Number: {sort}, Name: {name_cn or name}, Airdate: {airdate}" ) total = response.get("total", 0) results_text = f"Found {len(episodes)} episodes (Total: {total}).\n" + "---\n".join( formatted_results ) return results_text
• main.py:39-52 (schema)

  EpType IntEnum defines the episode types used as input parameter type for the get_episodes tool (episode_type: Optional[EpType]), providing schema/validation for episode filtering (0: Main Story, 1: SP, 2: OP, etc.).
  class EpType(IntEnum): """ 章节类型 0 = 本篇, 1 = 特别篇, 2 = OP, 3 = ED, 4 = 预告/宣传/广告, 5 = MAD, 6 = 其他 """ MAIN_STORY = 0 SP = 1 OP = 2 ED = 3 PV = 4 MAD = 5 OTHER = 6
• main.py:105-169 (helper)

  Shared helper function make_bangumi_request used by get_episodes to perform authenticated HTTP requests to the Bangumi API, handle errors, and return JSON responses.
  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}
• main.py:172-205 (helper)

  Shared helper function handle_api_error_response used by get_episodes to check API responses for errors and format error messages.
  def handle_api_error_response(response: Any) -> Optional[str]: """ Checks if the API response indicates an error and returns a formatted error message. Handles both dictionary-based errors and returns from make_bangumi_request on failure. """ # Check for error structure returned by make_bangumi_request on HTTPStatusError or RequestError if isinstance(response, dict) and ( "error" in response or "status_code" in response ): # This is an error dictionary created by our helper status_code = response.get("status_code", "N/A") error_msg = response.get("error", "Unknown error during request.") details = response.get("details", "") return f"Bangumi API Request Error (Status {status_code}): {error_msg}. Details: {details}".strip() # Check for error structure returned by Bangumi API itself (often dictionaries) # Safely check if the response is a dictionary before accessing its keys if isinstance(response, dict): if "title" in response and "description" in response: # This looks like a common Bangumi error response structure error_title = response.get("title", "API Error") error_description = response.get("description", "No description provided.") # The API might return a status code in the body too, or rely on HTTP status return f"Bangumi API Error: {error_title}. {error_description}".strip() # Check if it's a dictionary but *not* empty and *doesn't* look like a success response from list endpoints # Check for specific error fields if structure varies # Add more checks here if other error dictionary formats are observed # Example: if "message" in response and "code" in response: return f"API Error {response['code']}: {response['message']}" pass # If it's a dictionary but doesn't match known error formats, assume it's a valid data response for now # If it's not a dictionary, or it's a dictionary that doesn't match known error formats, assume it's not an error return None