#!/usr/bin/env python3
"""
Persona MCP Server (페르소나 전용)
"""
import asyncio
import sys
import os
import logging
from fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Optional, List, Dict
from dotenv import load_dotenv
# FastAPI는 선택적 (HTTP 엔드포인트 사용 시만 필요)
try:
from fastapi import FastAPI
FASTAPI_AVAILABLE = True
except ImportError:
FASTAPI_AVAILABLE = False
# FastAPI가 없어도 MCP는 작동함
# 페르소나 도구 import
from .tools.persona import (
get_basic_profile_questions,
check_basic_profile_complete,
update_basic_profile,
create_fgi_survey,
generate_next_question,
submit_fgi_response,
check_freshness,
predict_persona,
get_persona,
list_personas,
get_persona_survey_history,
get_statistics,
compare_personas,
analyze_preference,
get_trend_insights,
search_responses_by_keyword,
generate_sample_personas,
generate_sample_surveys_and_responses,
generate_all_sample_data
)
# DB 초기화
from .db.database import init_db
# .env 파일 로드
load_dotenv()
# FastAPI / FastMCP 앱 구성
if FASTAPI_AVAILABLE:
api = FastAPI()
else:
api = None # FastAPI가 없으면 HTTP 엔드포인트 사용 불가
mcp_logger = logging.getLogger("persona-mcp")
level = getattr(logging, os.environ.get("LOG_LEVEL", "INFO").upper(), logging.INFO)
mcp_logger.setLevel(level)
if not mcp_logger.handlers:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
mcp_logger.addHandler(handler)
mcp_logger.propagate = True
# 앱 시작 시 DB 초기화
init_db()
# 서버 시작 시 기본 프로필 확인
def check_initial_profile():
"""서버 시작 시 기본 프로필이 설정되어 있는지 확인"""
profile_check = check_basic_profile_complete()
if not profile_check.get("is_complete", False):
mcp_logger.warning("=" * 60)
mcp_logger.warning("⚠️ 기본 프로필이 설정되지 않았습니다!")
mcp_logger.warning("=" * 60)
mcp_logger.warning(profile_check.get("message", "기본 프로필을 설정해주세요."))
mcp_logger.warning("")
mcp_logger.warning("📋 다음 단계:")
mcp_logger.warning("1. get_basic_profile_questions_tool()로 질문 목록 조회")
mcp_logger.warning("2. 사용자에게 질문하고 답변 수집")
mcp_logger.warning("3. update_basic_profile_tool()로 기본 프로필 저장")
mcp_logger.warning("")
mcp_logger.warning("필수 필드: age_range, gender, location, occupation_category")
mcp_logger.warning("=" * 60)
return False
else:
mcp_logger.info(f"✅ 기본 프로필 확인 완료: {profile_check.get('message', '')}")
return True
# 서버 시작 시 기본 프로필 확인
check_initial_profile()
# FastMCP 객체 생성
mcp = FastMCP("persona-server")
# ==================== Request Models ====================
class UpdateProfileRequest(BaseModel):
persona_id: str = Field(..., description="페르소나 ID")
profile_data: Dict = Field(..., description="기본 프로필 데이터")
class CreateSurveyRequest(BaseModel):
category: str = Field(..., description="설문 카테고리 (technology, lifestyle, culture, fashion, food, general)")
persona_id: Optional[str] = Field(None, description="페르소나 ID (선택사항)")
class NextQuestionRequest(BaseModel):
survey_id: str = Field(..., description="설문 ID")
previous_qa: List[Dict] = Field(..., description="이전 질문/답변 쌍")
category: Optional[str] = Field("general", description="카테고리")
max_questions: Optional[int] = Field(None, description="최대 질문 수 (선택사항, None이면 무제한)")
class SubmitResponseRequest(BaseModel):
persona_id: str = Field(..., description="페르소나 ID")
survey_id: str = Field(..., description="설문 ID")
responses: List[Dict] = Field(..., description="응답 리스트")
class CheckFreshnessRequest(BaseModel):
persona_id: str = Field(..., description="페르소나 ID")
class PredictPersonaRequest(BaseModel):
responses: List[Dict] = Field(..., description="설문 응답 리스트")
user_id: Optional[str] = Field(None, description="사용자 ID (선택사항)")
class GetPersonaRequest(BaseModel):
persona_id: str = Field(..., description="페르소나 ID")
class ListPersonasRequest(BaseModel):
age_range: Optional[str] = Field(None, description="나이대 필터")
gender: Optional[str] = Field(None, description="성별 필터")
location: Optional[str] = Field(None, description="지역 필터")
occupation_category: Optional[str] = Field(None, description="직업 필터")
mbti_type: Optional[str] = Field(None, description="MBTI 유형 필터")
min_freshness: Optional[float] = Field(None, description="최소 신선도 점수")
limit: int = Field(50, description="최대 결과 수")
class GetSurveyHistoryRequest(BaseModel):
persona_id: str = Field(..., description="페르소나 ID")
class ComparePersonasRequest(BaseModel):
persona_id1: str = Field(..., description="첫 번째 페르소나 ID")
persona_id2: str = Field(..., description="두 번째 페르소나 ID")
class AnalyzePreferenceRequest(BaseModel):
query: str = Field(..., description="분석할 주제/제품 (예: '스티커 디자인', '키링', '제품 느낌')")
category: Optional[str] = Field(None, description="설문 카테고리 필터")
age_range: Optional[str] = Field(None, description="나이대 필터")
gender: Optional[str] = Field(None, description="성별 필터")
location: Optional[str] = Field(None, description="지역 필터")
limit: int = Field(20, description="분석할 응답 수 제한")
class GetTrendInsightsRequest(BaseModel):
topic: str = Field(..., description="분석할 주제 (예: '스티커 디자인', '키링', '제품 느낌')")
persona_filters: Optional[Dict] = Field(None, description="페르소나 필터 (age_range, gender, location 등)")
min_freshness: Optional[float] = Field(None, description="최소 신선도 점수")
class SearchResponsesRequest(BaseModel):
keyword: str = Field(..., description="검색할 키워드")
category: Optional[str] = Field(None, description="설문 카테고리 필터")
limit: int = Field(20, description="최대 결과 수")
class GenerateSamplePersonasRequest(BaseModel):
count: int = Field(50, description="생성할 페르소나 수")
use_llm: bool = Field(True, description="LLM 사용 여부 (더 자연스러운 프로필)")
class GenerateSampleSurveysRequest(BaseModel):
persona_count: int = Field(20, description="응답을 생성할 페르소나 수")
responses_per_persona: int = Field(2, description="페르소나당 응답 수")
use_llm: bool = Field(True, description="LLM 사용 여부 (더 자연스러운 응답)")
class GenerateAllSampleDataRequest(BaseModel):
persona_count: int = Field(50, description="생성할 페르소나 수")
responses_per_persona: int = Field(2, description="페르소나당 응답 수")
use_llm: bool = Field(True, description="LLM 사용 여부")
# ==================== Implementation Functions ====================
async def update_profile_impl(req: UpdateProfileRequest):
try:
return await asyncio.to_thread(update_basic_profile, req.persona_id, req.profile_data)
except Exception as e:
return {"error": f"프로필 업데이트 중 오류: {str(e)}"}
async def create_survey_impl(req: CreateSurveyRequest):
try:
# 기본 프로필 확인 (persona_id가 있으면 해당 페르소나, 없으면 전체 확인)
profile_check = await asyncio.to_thread(check_basic_profile_complete, req.persona_id)
if not profile_check.get("is_complete", False):
return {
"error": "기본 프로필이 설정되지 않았습니다.",
"message": profile_check.get("message", "먼저 기본 프로필을 설정해주세요."),
"missing_fields": profile_check.get("missing_fields", []),
"required_action": "get_basic_profile_questions_tool()로 질문 목록을 조회하고, update_basic_profile_tool()로 기본 프로필을 저장하세요."
}
return await asyncio.to_thread(create_fgi_survey, req.category, req.persona_id)
except Exception as e:
return {"error": f"설문 생성 중 오류: {str(e)}"}
async def next_question_impl(req: NextQuestionRequest):
try:
return await asyncio.to_thread(generate_next_question, req.survey_id, req.previous_qa, req.category, req.max_questions)
except Exception as e:
return {"error": f"다음 질문 생성 중 오류: {str(e)}"}
async def submit_response_impl(req: SubmitResponseRequest):
try:
# 기본 프로필 확인
profile_check = await asyncio.to_thread(check_basic_profile_complete, req.persona_id)
if not profile_check.get("is_complete", False):
return {
"error": "기본 프로필이 설정되지 않았습니다.",
"message": profile_check.get("message", "먼저 기본 프로필을 설정해주세요."),
"missing_fields": profile_check.get("missing_fields", []),
"required_action": "get_basic_profile_questions_tool()로 질문 목록을 조회하고, update_basic_profile_tool()로 기본 프로필을 저장하세요."
}
return await asyncio.to_thread(submit_fgi_response, req.persona_id, req.survey_id, req.responses)
except Exception as e:
return {"error": f"응답 제출 중 오류: {str(e)}"}
async def check_freshness_impl(req: CheckFreshnessRequest):
try:
return await asyncio.to_thread(check_freshness, req.persona_id)
except Exception as e:
return {"error": f"신선도 확인 중 오류: {str(e)}"}
async def predict_persona_impl(req: PredictPersonaRequest):
try:
return await asyncio.to_thread(predict_persona, req.responses, req.user_id)
except Exception as e:
return {"error": f"페르소나 예측 중 오류: {str(e)}"}
async def get_persona_impl(req: GetPersonaRequest):
try:
return await asyncio.to_thread(get_persona, req.persona_id)
except Exception as e:
return {"error": f"페르소나 조회 중 오류: {str(e)}"}
async def list_personas_impl(req: ListPersonasRequest):
try:
return await asyncio.to_thread(
list_personas,
req.age_range,
req.gender,
req.location,
req.occupation_category,
req.mbti_type,
req.min_freshness,
req.limit
)
except Exception as e:
return {"error": f"페르소나 목록 조회 중 오류: {str(e)}"}
async def get_survey_history_impl(req: GetSurveyHistoryRequest):
try:
return await asyncio.to_thread(get_persona_survey_history, req.persona_id)
except Exception as e:
return {"error": f"설문 히스토리 조회 중 오류: {str(e)}"}
async def get_statistics_impl():
try:
return await asyncio.to_thread(get_statistics)
except Exception as e:
return {"error": f"통계 조회 중 오류: {str(e)}"}
async def compare_personas_impl(req: ComparePersonasRequest):
try:
return await asyncio.to_thread(compare_personas, req.persona_id1, req.persona_id2)
except Exception as e:
return {"error": f"페르소나 비교 중 오류: {str(e)}"}
async def analyze_preference_impl(req: AnalyzePreferenceRequest):
try:
# news_mcp-main과 동일한 패턴: arguments는 함수 내부에서 None으로 전달
# (실제 arguments는 MCP 프로토콜을 통해 전달됨)
return await asyncio.to_thread(
analyze_preference,
req.query,
req.category,
req.age_range,
req.gender,
req.location,
req.limit,
None # arguments는 함수 내부에서 처리
)
except Exception as e:
return {"error": f"선호도 분석 중 오류: {str(e)}"}
async def get_trend_insights_impl(req: GetTrendInsightsRequest):
try:
# news_mcp-main과 동일한 패턴
return await asyncio.to_thread(
get_trend_insights,
req.topic,
req.persona_filters,
req.min_freshness,
None # arguments는 함수 내부에서 처리
)
except Exception as e:
return {"error": f"트렌드 인사이트 분석 중 오류: {str(e)}"}
async def search_responses_impl(req: SearchResponsesRequest):
try:
# news_mcp-main과 동일한 패턴
return await asyncio.to_thread(
search_responses_by_keyword,
req.keyword,
req.category,
req.limit,
None # arguments는 함수 내부에서 처리
)
except Exception as e:
return {"error": f"응답 검색 중 오류: {str(e)}"}
async def generate_sample_personas_impl(req: GenerateSamplePersonasRequest):
try:
return await asyncio.to_thread(
generate_sample_personas,
req.count,
req.use_llm
)
except Exception as e:
return {"error": f"샘플 페르소나 생성 중 오류: {str(e)}"}
async def generate_sample_surveys_impl(req: GenerateSampleSurveysRequest):
try:
return await asyncio.to_thread(
generate_sample_surveys_and_responses,
req.persona_count,
req.responses_per_persona,
req.use_llm
)
except Exception as e:
return {"error": f"샘플 설문 생성 중 오류: {str(e)}"}
async def generate_all_sample_data_impl(req: GenerateAllSampleDataRequest):
try:
return await asyncio.to_thread(
generate_all_sample_data,
req.persona_count,
req.responses_per_persona,
req.use_llm
)
except Exception as e:
return {"error": f"샘플 데이터 생성 중 오류: {str(e)}"}
async def health_impl():
"""실제 서비스 상태 확인 구현"""
groq_key = os.environ.get("GROQ_API_KEY", "")
groq_key_status = "설정됨" if groq_key else "설정되지 않음"
# 기본 프로필 상태 확인
profile_check = await asyncio.to_thread(check_basic_profile_complete)
status = "ok"
if not profile_check.get("is_complete", False):
status = "warning" # 기본 프로필이 없으면 경고 상태
return {
"status": status,
"server": "persona-server",
"environment": {
"groq_api_key": groq_key_status,
"api_key_preview": groq_key[:10] + "..." if groq_key else "None"
},
"database": "sqlite:///data/fgi.db",
"basic_profile": {
"is_complete": profile_check.get("is_complete", False),
"message": profile_check.get("message", ""),
"missing_fields": profile_check.get("missing_fields", [])
}
}
async def get_tool_definitions_impl():
"""실제 도구 정의 구현"""
tools = [
{
"name": "health",
"description": "서비스 상태 확인",
"parameters": {"type": "object", "properties": {}, "required": []}
},
{
"name": "update_basic_profile_tool",
"description": "페르소나의 기본 프로필을 업데이트합니다.",
"parameters": {
"type": "object",
"properties": {
"persona_id": {"type": "string", "description": "페르소나 ID"},
"profile_data": {"type": "object", "description": "기본 프로필 데이터"}
},
"required": ["persona_id", "profile_data"]
}
},
{
"name": "create_fgi_survey_tool",
"description": "카테고리별 오프닝 질문으로 FGI 설문을 시작합니다.",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string", "description": "설문 카테고리 (technology, lifestyle, culture, fashion, food, general)"},
"persona_id": {"type": "string", "description": "페르소나 ID (선택사항)"}
},
"required": ["category"]
}
},
{
"name": "analyze_preference_tool",
"description": "특정 제품/주제에 대한 페르소나들의 선호도를 분석합니다.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "분석할 주제/제품 (예: '스티커 디자인', '키링')"},
"category": {"type": "string", "description": "설문 카테고리 필터"},
"age_range": {"type": "string", "description": "나이대 필터"},
"gender": {"type": "string", "description": "성별 필터"},
"location": {"type": "string", "description": "지역 필터"},
"limit": {"type": "integer", "description": "분석할 응답 수 제한", "default": 20}
},
"required": ["query"]
}
},
{
"name": "get_trend_insights_tool",
"description": "특정 주제에 대한 트렌드 인사이트를 제공합니다.",
"parameters": {
"type": "object",
"properties": {
"topic": {"type": "string", "description": "분석할 주제 (예: '스티커 디자인', '키링')"},
"persona_filters": {"type": "object", "description": "페르소나 필터"},
"min_freshness": {"type": "number", "description": "최소 신선도 점수"}
},
"required": ["topic"]
}
},
{
"name": "search_responses_tool",
"description": "AI 기반 의미 검색으로 특정 주제와 관련된 응답을 찾습니다.",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "검색할 주제/키워드"},
"category": {"type": "string", "description": "설문 카테고리 필터"},
"limit": {"type": "integer", "description": "최대 결과 수", "default": 20}
},
"required": ["keyword"]
}
},
{
"name": "get_statistics_tool",
"description": "전체 통계 정보를 조회합니다.",
"parameters": {"type": "object", "properties": {}, "required": []}
}
]
return {"tools": tools}
# HTTP 엔드포인트: 헬스체크 (FastAPI 사용 가능한 경우만)
if FASTAPI_AVAILABLE and api:
@api.get("/health")
async def health_check():
return await health_impl()
# HTTP 엔드포인트: 도구 정의 조회
@api.get("/tools")
async def get_tools_http():
definitions = await get_tool_definitions_impl()
return definitions.get("tools", [])
# ==================== MCP 도구 등록 ====================
@mcp.tool()
async def health():
"""서비스 상태 확인"""
return await health_impl()
@mcp.tool()
async def get_tool_definitions():
"""MCP 도구 정의를 JSON 형식으로 제공합니다."""
return await get_tool_definitions_impl()
@mcp.tool()
async def get_basic_profile_questions_tool():
"""
기본 프로필 질문 목록을 조회합니다.
페르소나의 기본 정보(나이대, 성별, 지역, 직업, 소득)를 수집하기 위한 질문 목록을 반환합니다.
각 질문은 선택형(choice)이며, 옵션 리스트가 포함되어 있습니다.
반환된 질문 목록을 사용하여 사용자에게 질문하고, 수집한 답변을 update_basic_profile_tool로 저장하세요.
"""
return await asyncio.to_thread(get_basic_profile_questions)
@mcp.tool()
async def check_basic_profile_tool(persona_id: Optional[str] = None):
"""
기본 프로필이 완전한지 확인합니다.
필수 필드: age_range, gender, location, occupation_category
선택 필드: income_range
persona_id를 제공하면 특정 페르소나를 확인하고, 없으면 모든 페르소나를 확인합니다.
"""
return await asyncio.to_thread(check_basic_profile_complete, persona_id)
@mcp.tool()
async def update_basic_profile_tool(persona_id: str, profile_data: Dict):
"""
페르소나의 기본 프로필을 업데이트합니다.
profile_data 형식: {
"age_range": "20대 후반",
"gender": "남성",
"location": "서울",
"occupation_category": "IT/개발",
"income_range": "300-500만원" (선택)
}
"""
req = UpdateProfileRequest(persona_id=persona_id, profile_data=profile_data)
return await update_profile_impl(req)
@mcp.tool()
async def create_fgi_survey_tool(category: str, persona_id: Optional[str] = None):
"""
카테고리별 오프닝 질문으로 FGI 설문을 시작합니다.
카테고리: technology, lifestyle, culture, fashion, food, general
"""
req = CreateSurveyRequest(category=category, persona_id=persona_id)
return await create_survey_impl(req)
@mcp.tool()
async def generate_next_question_tool(survey_id: str, previous_qa: List[Dict], category: str = "general", max_questions: Optional[int] = None):
"""
이전 답변을 분석해서 다음 질문을 동적으로 생성합니다.
**최소한의 질문으로 최대한의 정보를 수집**하는 것이 목표입니다.
previous_qa 형식: [{"question": "질문 내용", "answer": "답변 내용"}]
max_questions: 최대 질문 수 (선택사항, 예: 5-10개로 제한 가능)
반환값에 info_completeness(정보 수집 완료도)와 is_final(종료 여부)가 포함됩니다.
"""
req = NextQuestionRequest(survey_id=survey_id, previous_qa=previous_qa, category=category, max_questions=max_questions)
return await next_question_impl(req)
@mcp.tool()
async def submit_fgi_response_tool(persona_id: str, survey_id: str, responses: List[Dict]):
"""
FGI 설문 응답을 제출하고 페르소나의 신선도 점수를 업데이트합니다.
responses 형식: [{"question_id": "q1", "answer": "답변 내용"}]
"""
req = SubmitResponseRequest(persona_id=persona_id, survey_id=survey_id, responses=responses)
return await submit_response_impl(req)
@mcp.tool()
async def check_freshness_tool(persona_id: str):
"""
페르소나의 현재 신선도 점수를 확인합니다.
신선도는 시간이 지남에 따라 감소하며, 설문 응답으로 증가합니다.
"""
req = CheckFreshnessRequest(persona_id=persona_id)
return await check_freshness_impl(req)
@mcp.tool()
async def predict_persona_tool(responses: List[Dict], user_id: Optional[str] = None):
"""
설문 응답을 분석하여 사용자의 페르소나를 예측합니다.
AI 기반 분석으로 연령대, 관심사, 성격, 가치관, MBTI 유형 등을 파악합니다.
MBTI 스타일 성향 분석(외향/내향, 감각/직관, 사고/감정, 판단/인식)도 포함됩니다.
"""
req = PredictPersonaRequest(responses=responses, user_id=user_id)
return await predict_persona_impl(req)
@mcp.tool()
async def get_persona_tool(persona_id: str):
"""
특정 페르소나의 상세 정보를 조회합니다.
기본 프로필, 페르소나 프로필, MBTI 분석, 통계 정보를 포함합니다.
"""
req = GetPersonaRequest(persona_id=persona_id)
return await get_persona_impl(req)
@mcp.tool()
async def list_personas_tool(
age_range: Optional[str] = None,
gender: Optional[str] = None,
location: Optional[str] = None,
occupation_category: Optional[str] = None,
mbti_type: Optional[str] = None,
min_freshness: Optional[float] = None,
limit: int = 50
):
"""
페르소나 목록을 조회합니다. 다양한 필터 조건으로 검색할 수 있습니다.
필터: age_range, gender, location, occupation_category, mbti_type, min_freshness
"""
req = ListPersonasRequest(
age_range=age_range,
gender=gender,
location=location,
occupation_category=occupation_category,
mbti_type=mbti_type,
min_freshness=min_freshness,
limit=limit
)
return await list_personas_impl(req)
@mcp.tool()
async def get_persona_survey_history_tool(persona_id: str):
"""
특정 페르소나의 설문 히스토리를 조회합니다.
모든 설문 응답과 상세 정보를 포함합니다.
"""
req = GetSurveyHistoryRequest(persona_id=persona_id)
return await get_survey_history_impl(req)
@mcp.tool()
async def get_statistics_tool():
"""
전체 통계 정보를 조회합니다.
페르소나 통계, 설문 통계, MBTI 분포, 지역 분포, 직업 분포 등을 포함합니다.
"""
return await get_statistics_impl()
@mcp.tool()
async def compare_personas_tool(persona_id1: str, persona_id2: str):
"""
두 페르소나를 비교합니다.
기본 프로필, MBTI, 관심사, 성격 특성 등의 유사점과 차이점을 분석합니다.
"""
req = ComparePersonasRequest(persona_id1=persona_id1, persona_id2=persona_id2)
return await compare_personas_impl(req)
@mcp.tool()
async def analyze_preference_tool(
query: str,
category: Optional[str] = None,
age_range: Optional[str] = None,
gender: Optional[str] = None,
location: Optional[str] = None,
limit: int = 20
):
"""
특정 제품/주제에 대한 페르소나들의 선호도를 분석합니다.
비즈니스 질문 예시:
- "스티커 디자인" → 요즘 어떤 스티커 디자인을 좋아하는지
- "키링" → 어떤 키링을 좋아하는지
- "제품 느낌" → 어떤 느낌의 제품을 좋아하는지
필터를 통해 특정 타겟 그룹의 선호도만 분석할 수 있습니다.
"""
req = AnalyzePreferenceRequest(
query=query,
category=category,
age_range=age_range,
gender=gender,
location=location,
limit=limit
)
return await analyze_preference_impl(req)
@mcp.tool()
async def get_trend_insights_tool(
topic: str,
persona_filters: Optional[Dict] = None,
min_freshness: Optional[float] = None
):
"""
특정 주제에 대한 트렌드 인사이트를 제공합니다.
비즈니스 질문 예시:
- "스티커 디자인" → 요즘 인기 있는 스티커 디자인 트렌드
- "키링" → 인기 있는 키링 스타일
- "제품 느낌" → 요즘 선호하는 제품 느낌
페르소나 필터를 통해 특정 타겟 그룹의 트렌드만 분석할 수 있습니다.
"""
req = GetTrendInsightsRequest(
topic=topic,
persona_filters=persona_filters,
min_freshness=min_freshness
)
return await get_trend_insights_impl(req)
@mcp.tool()
async def search_responses_tool(
keyword: str,
category: Optional[str] = None,
limit: int = 20
):
"""
특정 키워드가 포함된 응답을 검색합니다.
예: "스티커", "키링", "디자인" 등의 키워드로 관련 응답을 찾을 수 있습니다.
"""
req = SearchResponsesRequest(keyword=keyword, category=category, limit=limit)
return await search_responses_impl(req)
@mcp.tool()
async def generate_sample_personas_tool(count: int = 50, use_llm: bool = True):
"""
샘플 페르소나를 생성합니다. 비즈니스 데모 및 테스트용입니다.
다양한 프로필의 페르소나를 자동으로 생성하여 즉시 테스트할 수 있습니다.
use_llm=True면 더 자연스러운 프로필을 생성하지만 비용이 발생합니다.
"""
req = GenerateSamplePersonasRequest(count=count, use_llm=use_llm)
return await generate_sample_personas_impl(req)
@mcp.tool()
async def generate_sample_surveys_tool(
persona_count: int = 20,
responses_per_persona: int = 2,
use_llm: bool = True
):
"""
샘플 설문 및 응답을 생성합니다. 비즈니스 데모 및 테스트용입니다.
기존 페르소나들에 대해 다양한 카테고리의 설문과 자연스러운 응답을 생성합니다.
"""
req = GenerateSampleSurveysRequest(
persona_count=persona_count,
responses_per_persona=responses_per_persona,
use_llm=use_llm
)
return await generate_sample_surveys_impl(req)
@mcp.tool()
async def generate_all_sample_data_tool(
persona_count: int = 50,
responses_per_persona: int = 2,
use_llm: bool = True
):
"""
전체 샘플 데이터를 한 번에 생성합니다. (페르소나 + 설문 + 응답)
비즈니스 데모 및 테스트를 위해 즉시 사용 가능한 데이터를 생성합니다.
use_llm=False로 설정하면 템플릿 기반으로 빠르게 생성됩니다 (비용 없음).
"""
req = GenerateAllSampleDataRequest(
persona_count=persona_count,
responses_per_persona=responses_per_persona,
use_llm=use_llm
)
return await generate_all_sample_data_impl(req)
# ==================== 서버 실행 ====================
async def main():
print("Persona MCP Server starting...", file=sys.stderr)
print("Tools: update_profile, create_survey, generate_next_question, submit_response, check_freshness, predict_persona, get_persona, list_personas, get_survey_history, get_statistics, compare_personas, analyze_preference, get_trend_insights, search_responses, generate_sample_personas, generate_sample_surveys, generate_all_sample_data", file=sys.stderr)
try:
await mcp.run_stdio_async()
except Exception as e:
print(f"Server error: {e}", file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
raise
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Server stopped by user", file=sys.stderr)
except Exception as e:
print(f"Server failed: {e}", file=sys.stderr)
sys.exit(1)
