MCP Chat Analysis Server
- src
- mcp_chat_analysis
from typing import List, Dict, Optional, Union
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field
class ConversationFormat(str, Enum):
"""Supported conversation import formats"""
OPENAI_NATIVE = "openai_native"
HTML = "html"
MARKDOWN = "markdown"
JSON = "json"
class ConversationData(BaseModel):
"""Input schema for conversation import"""
source_path: str = Field(
description="Path to the chat export file"
)
format: ConversationFormat = Field(
description="Format of the chat export"
)
metadata: Optional[Dict] = Field(
default=None,
description="Additional metadata for the import"
)
class SearchFilter(BaseModel):
"""Filter criteria for semantic search"""
field: str
operator: str
value: Union[str, int, float, bool, List]
class SearchQuery(BaseModel):
"""Input schema for semantic search"""
query: str = Field(
description="Search query text"
)
limit: int = Field(
default=10,
ge=1,
le=100,
description="Maximum number of results"
)
filters: Optional[List[SearchFilter]] = Field(
default=None,
description="Optional filters for search results"
)
min_score: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description="Minimum similarity score"
)
class MetricType(str, Enum):
"""Types of conversation metrics"""
MESSAGE_FREQUENCY = "message_frequency"
RESPONSE_TIMES = "response_times"
TOPIC_DIVERSITY = "topic_diversity"
CONVERSATION_DEPTH = "conversation_depth"
INTERACTION_PATTERNS = "interaction_patterns"
class MetricsRequest(BaseModel):
"""Input schema for metrics analysis"""
conversation_id: str = Field(
description="ID of the conversation to analyze"
)
metrics: List[MetricType] = Field(
description="Metrics to analyze"
)
time_window: Optional[str] = Field(
default=None,
description="Time window for analysis (e.g., '1d', '1w', '1m')"
)
class ConceptRequest(BaseModel):
"""Input schema for concept extraction"""
conversation_id: str = Field(
description="ID of the conversation to analyze"
)
min_relevance: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description="Minimum relevance score for concepts"
)
max_concepts: int = Field(
default=10,
ge=1,
description="Maximum number of concepts to extract"
)
class Message(BaseModel):
"""Chat message data model"""
id: str
content: str
role: str
timestamp: datetime
parent_id: Optional[str] = None
metadata: Dict = Field(default_factory=dict)
class Conversation(BaseModel):
"""Conversation data model"""
id: str
title: Optional[str]
messages: List[Message]
create_time: datetime
update_time: Optional[datetime]
metadata: Dict = Field(default_factory=dict)
class SearchResult(BaseModel):
"""Semantic search result"""
message: Message
conversation: Conversation
score: float
context: Dict = Field(default_factory=dict)
class ConceptNode(BaseModel):
"""Extracted concept node"""
id: str
name: str
relevance: float
frequency: int
first_occurrence: datetime
last_occurrence: datetime
related_concepts: List[str] = Field(default_factory=list)
class MetricsResult(BaseModel):
"""Conversation metrics result"""
conversation_id: str
time_window: Optional[str]
metrics: Dict[MetricType, Dict]
generated_at: datetime = Field(default_factory=datetime.now)