#!/usr/bin/env python3
"""Cyphers MCP Server - Neople Cyphers Open API를 MCP Tools로 노출하는 서버"""
import asyncio
import json
import os
import sys
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
TextContent,
Tool,
)
from .api_client import CyphersAPIClient
from .types import CyphersAPIConfig, CyphersAPIError
# MCP 서버 인스턴스
server = Server("cyphers-mcp-server")
# API 클라이언트 (초기화는 main에서)
api_client: CyphersAPIClient | None = None
# ========== Tool 정의 ==========
TOOLS = [
# Player Tools
Tool(
name="cy_players_search",
description="닉네임으로 플레이어를 검색합니다.",
inputSchema={
"type": "object",
"properties": {
"nickname": {
"type": "string",
"description": "검색할 플레이어 닉네임",
},
"wordType": {
"type": "string",
"enum": ["match", "full"],
"description": "검색 타입: match(부분일치) 또는 full(완전일치)",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한 (기본값: 10)",
},
},
"required": ["nickname"],
},
),
Tool(
name="cy_players_get",
description="플레이어 ID로 플레이어 상세 정보를 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"playerId": {
"type": "string",
"description": "플레이어 ID",
},
},
"required": ["playerId"],
},
),
Tool(
name="cy_players_matches",
description="플레이어의 매칭 기록을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"playerId": {
"type": "string",
"description": "플레이어 ID",
},
"gameTypeId": {
"type": "string",
"description": "게임 타입 ID",
},
"startDate": {
"type": "string",
"description": "시작 날짜 (YYYYMMDD 형식, 최대 90일)",
},
"endDate": {
"type": "string",
"description": "종료 날짜 (YYYYMMDD 형식, 최대 90일)",
},
"next": {
"type": "string",
"description": "페이지네이션을 위한 next 토큰",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한",
},
},
"required": ["playerId"],
},
),
# Match Tools
Tool(
name="cy_matches_get",
description="매치 ID로 매칭 상세 정보를 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"matchId": {
"type": "string",
"description": "매치 ID",
},
},
"required": ["matchId"],
},
),
# Ranking Tools
Tool(
name="cy_ranking_ratingpoint",
description="통합 랭킹(레이팅 포인트)을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"playerId": {
"type": "string",
"description": "플레이어 ID",
},
"nickname": {
"type": "string",
"description": "플레이어 닉네임",
},
"offset": {
"type": "integer",
"minimum": 0,
"description": "오프셋",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한",
},
},
"required": [],
},
),
Tool(
name="cy_ranking_characters",
description="캐릭터별 랭킹을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"characterId": {
"type": "string",
"description": "캐릭터 ID",
},
"rankingType": {
"type": "string",
"description": "랭킹 타입",
},
"offset": {
"type": "integer",
"minimum": 0,
"description": "오프셋",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한",
},
},
"required": ["characterId", "rankingType"],
},
),
Tool(
name="cy_ranking_tsj",
description="투신전 랭킹을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"tsjType": {
"type": "string",
"description": "투신전 타입",
},
"offset": {
"type": "integer",
"minimum": 0,
"description": "오프셋",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한",
},
},
"required": ["tsjType"],
},
),
# Item Tools
Tool(
name="cy_battleitems_search",
description="아이템을 검색합니다.",
inputSchema={
"type": "object",
"properties": {
"itemName": {
"type": "string",
"description": "아이템 이름",
},
"characterId": {
"type": "string",
"description": "캐릭터 ID",
},
"slotCode": {
"type": "string",
"description": "슬롯 코드",
},
"rarityCode": {
"type": "string",
"description": "레어리티 코드",
},
"seasonCode": {
"type": "string",
"description": "시즌 코드",
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"description": "결과 개수 제한",
},
},
"required": [],
},
),
Tool(
name="cy_battleitems_get",
description="아이템 ID로 아이템 상세 정보를 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"itemId": {
"type": "string",
"description": "아이템 ID",
},
},
"required": ["itemId"],
},
),
Tool(
name="cy_battleitems_multi_get",
description="여러 아이템의 상세 정보를 한 번에 조회합니다 (최대 30개).",
inputSchema={
"type": "object",
"properties": {
"itemIds": {
"type": "array",
"items": {"type": "string"},
"minItems": 1,
"maxItems": 30,
"description": "아이템 ID 배열 (최대 30개)",
},
},
"required": ["itemIds"],
},
),
# Character Tools
Tool(
name="cy_characters_list",
description="모든 캐릭터 목록을 조회합니다.",
inputSchema={
"type": "object",
"properties": {},
"required": [],
},
),
# Image URL Tools
Tool(
name="cy_images_character_url",
description="캐릭터 이미지 URL을 생성합니다.",
inputSchema={
"type": "object",
"properties": {
"characterId": {
"type": "string",
"description": "캐릭터 ID",
},
"zoom": {
"type": "integer",
"minimum": 1,
"maximum": 3,
"description": "줌 레벨 (1-3)",
},
},
"required": ["characterId"],
},
),
Tool(
name="cy_images_item_url",
description="아이템 이미지 URL을 생성합니다.",
inputSchema={
"type": "object",
"properties": {
"itemId": {
"type": "string",
"description": "아이템 ID",
},
},
"required": ["itemId"],
},
),
]
# ========== MCP Handlers ==========
@server.list_tools()
async def list_tools() -> list[Tool]:
"""사용 가능한 Tool 목록 반환"""
return TOOLS
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Tool 실행"""
global api_client
if api_client is None:
return [TextContent(
type="text",
text=json.dumps({"error": {"code": "NOT_INITIALIZED", "message": "API client not initialized"}}, indent=2),
)]
try:
result = await execute_tool(name, arguments)
return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))]
except CyphersAPIError as e:
return [TextContent(type="text", text=json.dumps(e.to_dict(), indent=2, ensure_ascii=False))]
except Exception as e:
return [TextContent(
type="text",
text=json.dumps({"error": {"code": "INTERNAL_ERROR", "message": str(e)}}, indent=2),
)]
async def execute_tool(name: str, args: dict[str, Any]) -> dict[str, Any]:
"""Tool 실행 로직"""
global api_client
if api_client is None:
raise CyphersAPIError("API client not initialized", error_code="NOT_INITIALIZED")
# Player Tools
if name == "cy_players_search":
response = await api_client.search_players(
nickname=args["nickname"],
word_type=args.get("wordType"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_players_get":
response = await api_client.get_player(args["playerId"])
return {
"player": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_players_matches":
response = await api_client.get_player_matches(
player_id=args["playerId"],
game_type_id=args.get("gameTypeId"),
start_date=args.get("startDate"),
end_date=args.get("endDate"),
next_token=args.get("next"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
# Match Tools
elif name == "cy_matches_get":
response = await api_client.get_match(args["matchId"])
return {
"match": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
# Ranking Tools
elif name == "cy_ranking_ratingpoint":
response = await api_client.get_ranking_rating_point(
player_id=args.get("playerId"),
nickname=args.get("nickname"),
offset=args.get("offset"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_ranking_characters":
response = await api_client.get_ranking_characters(
character_id=args["characterId"],
ranking_type=args["rankingType"],
offset=args.get("offset"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_ranking_tsj":
response = await api_client.get_ranking_tsj(
tsj_type=args["tsjType"],
offset=args.get("offset"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
# Item Tools
elif name == "cy_battleitems_search":
response = await api_client.search_items(
item_name=args.get("itemName"),
character_id=args.get("characterId"),
slot_code=args.get("slotCode"),
rarity_code=args.get("rarityCode"),
season_code=args.get("seasonCode"),
limit=args.get("limit"),
)
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_battleitems_get":
response = await api_client.get_item(args["itemId"])
return {
"item": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
elif name == "cy_battleitems_multi_get":
response = await api_client.get_multi_items(args["itemIds"])
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
# Character Tools
elif name == "cy_characters_list":
response = await api_client.get_characters()
return {
"rows": response.data,
"meta": {
"request_id": response.request_id,
"cached": response.cached,
"rate_limit": response.rate_limit,
},
}
# Image URL Tools
elif name == "cy_images_character_url":
character_id = args["characterId"]
zoom = args.get("zoom")
base_url = "https://img-api.neople.co.kr/cy/characters"
zoom_param = f"?zoom={zoom}" if zoom else ""
url = f"{base_url}/{character_id}{zoom_param}"
return {
"url": url,
"meta": {
"request_id": "local",
"cached": False,
},
}
elif name == "cy_images_item_url":
item_id = args["itemId"]
url = f"https://img-api.neople.co.kr/cy/items/{item_id}"
return {
"url": url,
"meta": {
"request_id": "local",
"cached": False,
},
}
else:
raise CyphersAPIError(
f"Unknown tool: {name}",
error_code="UNKNOWN_TOOL",
)
async def run_server():
"""MCP 서버 실행"""
global api_client
# API 키 확인
api_key = os.environ.get("CYPHERS_API_KEY")
if not api_key:
print("Error: CYPHERS_API_KEY environment variable is required", file=sys.stderr)
sys.exit(1)
# API 클라이언트 초기화
api_client = CyphersAPIClient(CyphersAPIConfig(api_key=api_key))
print(f"Cyphers MCP Server running on stdio", file=sys.stderr)
print(f"API Key configured: {api_key[:8]}...", file=sys.stderr)
try:
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
finally:
if api_client:
await api_client.close()
def main():
"""메인 진입점"""
asyncio.run(run_server())
if __name__ == "__main__":
main()
