"""MCP Server for ai.card system"""
from typing import Optional, List, Dict, Any
from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from pathlib import Path
import logging
from app.core.config import settings
from app.db.base import get_session
from app.models.card import Card, CardRarity, CardDrawResult
from app.repositories.card import CardRepository, UniqueCardRepository
from app.repositories.user import UserRepository
from app.services.gacha import GachaService
# from app.services.card_sync import CardSyncService # Temporarily disabled
logger = logging.getLogger(__name__)
class AICardMcpServer:
"""MCP Server that exposes ai.card functionality to AI assistants"""
def __init__(self, enable_mcp: bool = True):
self.enable_mcp = enable_mcp
# Create FastAPI app
self.app = FastAPI(
title="AI.Card - Card Game System",
description="MCP server for ai.card system",
version=settings.app_version
)
# Create MCP server with FastAPI app
self.server = None
if enable_mcp:
self.server = FastMCP("aicard")
self._register_mcp_tools()
def get_app(self) -> FastAPI:
"""Get the FastAPI app instance"""
return self.app
def _register_mcp_tools(self):
"""Register all MCP tools"""
@self.app.get("/get_user_cards", operation_id="get_user_cards")
async def get_user_cards(
did: str,
limit: int = 10,
session: AsyncSession = Depends(get_session)
) -> List[Dict[str, Any]]:
"""Get all cards owned by a user"""
try:
user_repo = UserRepository(session)
card_repo = CardRepository(session)
# Get user
user = await user_repo.get_by_did(did)
if not user:
return []
# Get user cards
user_cards = await card_repo.get_user_cards(user.id, limit=limit)
return [
{
"id": card.card_id,
"cp": card.cp,
"status": card.status,
"skill": card.skill,
"owner_did": did,
"obtained_at": card.obtained_at.isoformat(),
"is_unique": card.is_unique,
"unique_id": str(card.unique_id) if card.unique_id else None
}
for card in user_cards
]
except Exception as e:
logger.error(f"Error getting user cards: {e}")
return []
@self.app.post("/draw_card", operation_id="draw_card")
async def draw_card(
did: str,
is_paid: bool = False,
session: AsyncSession = Depends(get_session)
) -> Dict[str, Any]:
"""Draw a new card (gacha) for user"""
try:
gacha_service = GachaService(session)
# Draw card
card, is_unique = await gacha_service.draw_card(did, is_paid)
await session.commit()
return {
"success": True,
"card": {
"id": card.id,
"cp": card.cp,
"status": card.status,
"skill": card.skill,
"owner_did": card.owner_did,
"obtained_at": card.obtained_at.isoformat(),
"is_unique": card.is_unique,
"unique_id": card.unique_id
},
"is_unique": is_unique,
"animation_type": "kira" if card.status in [CardRarity.KIRA, CardRarity.UNIQUE] else "normal"
}
except Exception as e:
logger.error(f"Error drawing card: {e}")
await session.rollback()
return {
"success": False,
"error": str(e)
}
@self.app.get("/get_card_details", operation_id="get_card_details")
async def get_card_details(
card_id: int,
session: AsyncSession = Depends(get_session)
) -> Dict[str, Any]:
"""Get detailed information about a card type"""
try:
# Get card info from gacha service
gacha_service = GachaService(session)
if card_id not in gacha_service.CARD_INFO:
return {"error": f"Card ID {card_id} not found"}
card_info = gacha_service.CARD_INFO[card_id]
# Get unique card availability
unique_repo = UniqueCardRepository(session)
is_unique_available = await unique_repo.is_card_available(card_id)
return {
"id": card_id,
"name": card_info["name"],
"base_cp_range": card_info["base_cp_range"],
"is_unique_available": is_unique_available,
"description": f"Card {card_id}: {card_info['name']}"
}
except Exception as e:
logger.error(f"Error getting card details: {e}")
return {"error": str(e)}
@self.app.post("/sync_cards_atproto", operation_id="sync_cards_atproto")
async def sync_cards_atproto(
did: str,
session: AsyncSession = Depends(get_session)
) -> Dict[str, str]:
"""Sync user's cards with atproto (temporarily disabled)"""
return {"status": "atproto sync temporarily disabled due to dependency issues"}
@self.app.get("/analyze_card_collection", operation_id="analyze_card_collection")
async def analyze_card_collection(
did: str,
session: AsyncSession = Depends(get_session)
) -> Dict[str, Any]:
"""Analyze user's card collection"""
try:
user_repo = UserRepository(session)
card_repo = CardRepository(session)
# Get user
user = await user_repo.get_by_did(did)
if not user:
return {
"total_cards": 0,
"rarity_distribution": {},
"message": "User not found"
}
# Get all user cards
user_cards = await card_repo.get_user_cards(user.id, limit=1000)
if not user_cards:
return {
"total_cards": 0,
"rarity_distribution": {},
"message": "No cards found"
}
# Analyze collection
rarity_count = {}
total_cp = 0
card_type_count = {}
for card in user_cards:
# Rarity distribution
rarity = card.status
rarity_count[rarity] = rarity_count.get(rarity, 0) + 1
# Total CP
total_cp += card.cp
# Card type distribution
card_type_count[card.card_id] = card_type_count.get(card.card_id, 0) + 1
# Find strongest card
strongest_card = max(user_cards, key=lambda x: x.cp)
return {
"total_cards": len(user_cards),
"rarity_distribution": rarity_count,
"card_type_distribution": card_type_count,
"average_cp": total_cp / len(user_cards) if user_cards else 0,
"total_cp": total_cp,
"strongest_card": {
"id": strongest_card.card_id,
"cp": strongest_card.cp,
"status": strongest_card.status,
"is_unique": strongest_card.is_unique
},
"unique_count": len([c for c in user_cards if c.is_unique])
}
except Exception as e:
logger.error(f"Error analyzing collection: {e}")
return {"error": str(e)}
@self.app.get("/get_unique_registry", operation_id="get_unique_registry")
async def get_unique_registry(
session: AsyncSession = Depends(get_session)
) -> Dict[str, Any]:
"""Get all registered unique cards"""
try:
unique_repo = UniqueCardRepository(session)
# Get all unique cards
unique_cards = await unique_repo.get_all_unique_cards()
# Get available unique card IDs
available_ids = await unique_repo.get_available_unique_cards()
return {
"registered_unique_cards": [
{
"card_id": card.card_id,
"unique_id": card.unique_id,
"owner_did": card.owner_did,
"obtained_at": card.obtained_at.isoformat()
}
for card in unique_cards
],
"available_unique_card_ids": available_ids,
"total_registered": len(unique_cards),
"total_available": len(available_ids)
}
except Exception as e:
logger.error(f"Error getting unique registry: {e}")
return {"error": str(e)}
@self.app.get("/get_gacha_stats", operation_id="get_gacha_stats")
async def get_gacha_stats(
session: AsyncSession = Depends(get_session)
) -> Dict[str, Any]:
"""Get gacha system statistics"""
try:
return {
"rarity_probabilities": {
"normal": f"{100 - settings.prob_rare}%",
"rare": f"{settings.prob_rare - settings.prob_super_rare}%",
"super_rare": f"{settings.prob_super_rare - settings.prob_kira}%",
"kira": f"{settings.prob_kira - settings.prob_unique}%",
"unique": f"{settings.prob_unique}%"
},
"total_card_types": 16,
"card_names": [info["name"] for info in GachaService.CARD_INFO.values()],
"system_info": {
"daily_limit": "1 free draw per day",
"paid_gacha": "Enhanced probabilities",
"unique_system": "First-come-first-served globally unique cards"
}
}
except Exception as e:
logger.error(f"Error getting gacha stats: {e}")
return {"error": str(e)}
# MCP server will be run separately, not here
def get_server(self) -> Optional[FastMCP]:
"""Get the FastAPI MCP server instance"""
return self.server
def get_app(self) -> FastAPI:
"""Get the FastAPI app instance"""
return self.app