#!/usr/bin/env python3
"""Murder Mystery MCP Server.
An MCP server that exposes a complete murder mystery game that any
MCP-compatible agent can play.
Usage:
python server.py
Or add to Claude Desktop config.
"""
import asyncio
import json
import logging
import os
import uuid
from typing import Any, Optional
from dotenv import load_dotenv
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
Resource,
ResourceTemplate,
)
from game.state import GameSession
from game.mystery_generator import generate_mystery_async
from game.oracle import MysteryOracle
from game.memory import GameMemory
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("murder-mystery-mcp")
# =============================================================================
# SERVER SETUP
# =============================================================================
server = Server("murder-mystery")
# Global game sessions (keyed by session_id)
sessions: dict[str, GameSession] = {}
def get_session(session_id: str) -> GameSession:
"""Get or create a game session."""
if session_id not in sessions:
sessions[session_id] = GameSession(session_id)
return sessions[session_id]
# =============================================================================
# TOOLS
# =============================================================================
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="start_game",
description="""Start a new murder mystery game.
Returns the premise, victim info, and list of suspects.
Configuration options:
- era: Category filter (e.g., 'Historical', 'Modern', 'Sci-Fi & Futuristic', 'Tech & Gaming')
- setting: Specific location (e.g., 'a 1920s speakeasy during Prohibition', 'a space station orbiting Mars')
- difficulty: Game difficulty ('Easy', 'Normal', 'Hard') - affects clue clarity and hints
- tone: Narrative style ('Classic Noir', 'Cozy Mystery', 'Darkly Comedic', 'Gothic Romance')""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for this game (optional, auto-generated if not provided)"
},
"era": {
"type": "string",
"description": "Era/category filter: 'Historical', 'Modern', 'Sci-Fi & Futuristic', 'Tech & Gaming', or 'Any'"
},
"setting": {
"type": "string",
"description": "Specific location/setting (e.g., 'a luxury cruise ship', 'a space station'). If not provided, randomly selected based on era."
},
"difficulty": {
"type": "string",
"enum": ["Easy", "Normal", "Hard"],
"description": "Game difficulty. Easy=clear clues, Hard=subtle clues with red herrings. Default: Normal"
},
"tone": {
"type": "string",
"description": "Narrative tone: 'Classic Noir', 'Cozy Mystery', 'Darkly Comedic', 'Flirty Noir', 'Gothic Romance', or 'Random'"
}
},
"required": []
}
),
Tool(
name="get_game_state",
description="""Get the current state of the game.
Returns suspects (with interrogation status), discovered clues,
searched locations, and investigation progress.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
}
},
"required": ["session_id"]
}
),
Tool(
name="interrogate_suspect",
description="""Talk to a suspect in the murder mystery.
Ask them questions, confront them with evidence, or build rapport.
Their responses depend on their emotional state (trust/nervousness).
The suspect remembers past conversations.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"suspect_name": {
"type": "string",
"description": "Name of the suspect to interrogate"
},
"question": {
"type": "string",
"description": "What you want to say or ask the suspect"
}
},
"required": ["session_id", "suspect_name", "question"]
}
),
Tool(
name="search_location",
description="""Search a location in the mystery for clues.
Locations become available as suspects reveal them during interrogation.
Searching reveals clues with their descriptions and significance.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"location": {
"type": "string",
"description": "Name of the location to search"
}
},
"required": ["session_id", "location"]
}
),
Tool(
name="make_accusation",
description="""Formally accuse a suspect of the murder.
This is a FINAL accusation - make sure you have evidence!
- You get 3 wrong accusations before losing
- To win: name the correct murderer with supporting evidence
- Evidence should include: discovered clues, alibi contradictions""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"suspect_name": {
"type": "string",
"description": "Name of the suspect you're accusing"
},
"evidence": {
"type": "string",
"description": "Your evidence and reasoning for the accusation"
}
},
"required": ["session_id", "suspect_name", "evidence"]
}
),
Tool(
name="search_memory",
description="""Search past conversations and clues using semantic search.
Use this to recall what suspects said about specific topics,
find statements related to alibis, or look up clue details.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"query": {
"type": "string",
"description": "What to search for (e.g., 'what did Marcus say about the night of the murder')"
},
"suspect_filter": {
"type": "string",
"description": "Optional: only search statements by this suspect"
}
},
"required": ["session_id", "query"]
}
),
Tool(
name="find_contradictions",
description="""Check if a suspect has contradicted themselves.
Analyzes all of a suspect's statements to find inconsistencies
that could indicate they're lying.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"suspect_name": {
"type": "string",
"description": "Name of the suspect to check for contradictions"
}
},
"required": ["session_id", "suspect_name"]
}
),
Tool(
name="get_timeline",
description="""Get the investigation timeline of discovered events.
Shows alibi claims, witness sightings, and clue discoveries
organized by time - useful for spotting inconsistencies.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
}
},
"required": ["session_id"]
}
),
# =================================================================
# IMAGE GENERATION TOOLS
# =================================================================
Tool(
name="generate_scene_image",
description="""Generate an image of a location or scene.
Creates a visual representation of a location in the mystery.
Returns the image URL or base64 data.
Style: 1990s point-and-click adventure game aesthetic.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"location": {
"type": "string",
"description": "Name or description of the location to visualize"
},
"include_clue": {
"type": "boolean",
"description": "Whether to include any clue found at this location"
}
},
"required": ["session_id", "location"]
}
),
Tool(
name="generate_portrait",
description="""Generate a portrait image of a suspect.
Creates a visual representation of a suspect based on their description.
Returns the image URL or base64 data.
Style: 1990s point-and-click adventure game character portrait.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
},
"suspect_name": {
"type": "string",
"description": "Name of the suspect to generate a portrait for"
}
},
"required": ["session_id", "suspect_name"]
}
),
Tool(
name="generate_title_card",
description="""Generate a title card image for the mystery.
Creates a dramatic title/opening scene image for the current mystery.
Returns the image URL or base64 data.
Style: 1990s point-and-click adventure game title screen.""",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "Session ID for the game"
}
},
"required": ["session_id"]
}
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "start_game":
return await handle_start_game(arguments)
elif name == "get_game_state":
return await handle_get_game_state(arguments)
elif name == "interrogate_suspect":
return await handle_interrogate_suspect(arguments)
elif name == "search_location":
return await handle_search_location(arguments)
elif name == "make_accusation":
return await handle_make_accusation(arguments)
elif name == "search_memory":
return await handle_search_memory(arguments)
elif name == "find_contradictions":
return await handle_find_contradictions(arguments)
elif name == "get_timeline":
return await handle_get_timeline(arguments)
# Image generation tools
elif name == "generate_scene_image":
return await handle_generate_scene_image(arguments)
elif name == "generate_portrait":
return await handle_generate_portrait(arguments)
elif name == "generate_title_card":
return await handle_generate_title_card(arguments)
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except Exception as e:
logger.exception(f"Error in tool {name}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
# =============================================================================
# TOOL HANDLERS
# =============================================================================
async def handle_start_game(args: dict) -> list[TextContent]:
"""Start a new murder mystery game."""
session_id = args.get("session_id") or str(uuid.uuid4())
era = args.get("era")
setting = args.get("setting")
difficulty = args.get("difficulty", "Normal")
tone = args.get("tone")
logger.info(f"Starting new game: session={session_id}, era={era}, setting={setting}, difficulty={difficulty}, tone={tone}")
session = get_session(session_id)
# Store difficulty for RAG settings
session.difficulty = difficulty
# Generate the mystery with all config options
mystery = await generate_mystery_async(
era=era,
setting=setting,
difficulty=difficulty,
tone=tone
)
# Initialize the Oracle with the truth
session.oracle.initialize(mystery)
# Create public view for the player
session.initialize_from_mystery(mystery)
# Build response
suspects_info = "\n".join([
f"- **{s.name}** ({s.role}): {s.personality}"
for s in session.public_suspects
])
response = f"""# 🔍 New Murder Mystery Started
## Session ID
`{session_id}` (use this for all subsequent commands)
## The Case
{session.setting}
## The Victim
**{session.victim_name}**: {session.victim_background}
## Suspects
{suspects_info}
## Available Actions
- Interrogate any suspect to learn more
- Locations will be revealed as you investigate
- Find clues to build your case
- Make an accusation when you're confident
Good luck, detective! 🕵️
"""
return [TextContent(type="text", text=response)]
async def handle_get_game_state(args: dict) -> list[TextContent]:
"""Get current game state."""
session_id = args.get("session_id")
if not session_id:
return [TextContent(type="text", text="Error: session_id is required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Build suspects status
suspects_status = []
for s in session.public_suspects:
status = "✅ Interrogated" if s.name in session.suspects_talked_to else "❓ Not talked to"
trust = session.get_suspect_trust(s.name)
nervousness = session.get_suspect_nervousness(s.name)
alibi_info = f" | Alibi: {s.alibi}" if s.name in session.suspects_talked_to else ""
suspects_status.append(
f"- **{s.name}** ({s.role}): {status}\n"
f" Trust: {trust}% | Nervousness: {nervousness}%{alibi_info}"
)
# Build clues
clues_text = "\n".join([f"- {c}" for c in session.discovered_clues]) or "None yet"
# Build locations
available_locs = "\n".join([
f"- {loc}" + (" ✅ Searched" if loc in session.searched_locations else "")
for loc in session.available_locations
]) or "None revealed yet - interrogate suspects to unlock locations"
response = f"""# 📊 Game State
## Session: `{session_id}`
## Suspects
{chr(10).join(suspects_status)}
## Discovered Clues ({len(session.discovered_clues)})
{clues_text}
## Available Locations
{available_locs}
## Progress
- Suspects interrogated: {len(session.suspects_talked_to)}/{len(session.public_suspects)}
- Clues found: {len(session.discovered_clues)}
- Locations searched: {len(session.searched_locations)}/{len(session.available_locations)}
- Wrong accusations: {session.wrong_accusations}/3
## Game Status
{"🎮 In Progress" if not session.game_over else ("🎉 WON!" if session.won else "💀 LOST")}
"""
return [TextContent(type="text", text=response)]
async def handle_interrogate_suspect(args: dict) -> list[TextContent]:
"""Interrogate a suspect."""
session_id = args.get("session_id")
suspect_name = args.get("suspect_name")
question = args.get("question")
if not all([session_id, suspect_name, question]):
return [TextContent(type="text", text="Error: session_id, suspect_name, and question are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Find the suspect
suspect = session.find_suspect(suspect_name)
if not suspect:
available = ", ".join([s.name for s in session.public_suspects])
return [TextContent(type="text", text=f"Suspect '{suspect_name}' not found. Available: {available}")]
# Get response from Oracle
result = await session.oracle.interrogate(
suspect_name=suspect.name,
question=question,
trust=session.get_suspect_trust(suspect.name),
nervousness=session.get_suspect_nervousness(suspect.name),
conversation_history=session.get_conversation_history(suspect.name),
)
# Update session state
session.record_conversation(suspect.name, question, result.response)
session.update_emotional_state(
suspect.name,
trust_delta=result.trust_delta,
nervousness_delta=result.nervousness_delta
)
# Handle reveals
if result.revealed_location:
session.unlock_location(result.revealed_location)
if result.revealed_secret:
session.mark_secret_revealed(suspect.name)
# Add to RAG memory
session.memory.add_conversation(suspect.name, question, result.response, session.turn)
# Build response
emotional_info = f"Trust: {session.get_suspect_trust(suspect.name)}% | Nervousness: {session.get_suspect_nervousness(suspect.name)}%"
location_note = ""
if result.revealed_location:
location_note = f"\n\n📍 **New Location Unlocked:** {result.revealed_location}"
secret_note = ""
if result.revealed_secret:
secret_note = f"\n\n🔓 **{suspect.name} revealed something important!**"
response = f"""# 🗣️ {suspect.name}
> {result.response}
---
*{emotional_info}*{location_note}{secret_note}
"""
return [TextContent(type="text", text=response)]
async def handle_search_location(args: dict) -> list[TextContent]:
"""Search a location for clues."""
session_id = args.get("session_id")
location = args.get("location")
if not all([session_id, location]):
return [TextContent(type="text", text="Error: session_id and location are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Check if location is available
matching_loc = session.find_location(location)
if not matching_loc:
available = ", ".join(session.available_locations) or "None - interrogate suspects first"
return [TextContent(type="text", text=f"Location '{location}' not available. Available: {available}")]
# Check if already searched
if matching_loc in session.searched_locations:
return [TextContent(type="text", text=f"You've already searched {matching_loc}. The clue you found there is in your notes.")]
# Search the location
result = await session.oracle.search_location(matching_loc)
# Update state
session.mark_location_searched(matching_loc)
if result.clue_found:
session.add_discovered_clue(result.clue_found)
session.memory.add_clue(
clue_id=result.clue_id,
description=result.clue_found,
location=matching_loc,
significance=result.significance,
turn=session.turn
)
response = f"""# 🔍 Searching: {matching_loc}
{result.scene_description}
"""
if result.clue_found:
response += f"""## 🔎 Clue Found!
**{result.clue_found}**
*Significance: {result.significance}*
"""
else:
response += "No significant clues found at this location."
return [TextContent(type="text", text=response)]
async def handle_make_accusation(args: dict) -> list[TextContent]:
"""Make a formal accusation."""
session_id = args.get("session_id")
suspect_name = args.get("suspect_name")
evidence = args.get("evidence", "")
if not all([session_id, suspect_name]):
return [TextContent(type="text", text="Error: session_id and suspect_name are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
if session.game_over:
return [TextContent(type="text", text="Game is already over.")]
# Find the suspect
suspect = session.find_suspect(suspect_name)
if not suspect:
available = ", ".join([s.name for s in session.public_suspects])
return [TextContent(type="text", text=f"Suspect '{suspect_name}' not found. Available: {available}")]
# Check accusation through Oracle
is_correct = session.oracle.check_accusation(suspect.name)
if is_correct:
session.game_over = True
session.won = True
murderer_name = session.oracle.get_murderer_name()
response = f"""# 🎉 CASE SOLVED!
**You correctly identified {murderer_name} as the murderer!**
Your evidence:
{evidence}
Congratulations, detective! Justice has been served. 🏆
"""
else:
session.wrong_accusations += 1
remaining = 3 - session.wrong_accusations
if remaining <= 0:
session.game_over = True
session.won = False
murderer_name = session.oracle.get_murderer_name()
response = f"""# 💀 GAME OVER
You've made 3 wrong accusations and have been removed from the case.
The actual murderer was **{murderer_name}**.
Better luck next time, detective.
"""
else:
response = f"""# ❌ Wrong Accusation
**{suspect.name}** is not the murderer.
You have **{remaining}** accusation(s) remaining before you're removed from the case.
Keep investigating!
"""
return [TextContent(type="text", text=response)]
async def handle_search_memory(args: dict) -> list[TextContent]:
"""Search past conversations and clues."""
session_id = args.get("session_id")
query = args.get("query")
suspect_filter = args.get("suspect_filter")
if not all([session_id, query]):
return [TextContent(type="text", text="Error: session_id and query are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Search memory
results = session.memory.search(query, suspect_filter=suspect_filter, k=5)
if not results:
return [TextContent(type="text", text=f"No relevant information found for: '{query}'")]
response = f"# 🔍 Memory Search: '{query}'\n\n"
for i, (text, metadata) in enumerate(results, 1):
source = metadata.get("suspect", metadata.get("location", "Unknown"))
turn = metadata.get("turn", "?")
response += f"**{i}. From {source} (Turn {turn}):**\n> {text}\n\n"
return [TextContent(type="text", text=response)]
async def handle_find_contradictions(args: dict) -> list[TextContent]:
"""Find contradictions in a suspect's statements."""
session_id = args.get("session_id")
suspect_name = args.get("suspect_name")
if not all([session_id, suspect_name]):
return [TextContent(type="text", text="Error: session_id and suspect_name are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Find the suspect
suspect = session.find_suspect(suspect_name)
if not suspect:
return [TextContent(type="text", text=f"Suspect '{suspect_name}' not found")]
# Check for contradictions
result = await session.memory.find_contradictions(suspect.name)
if not result.contradictions:
response = f"""# 🔍 Contradiction Check: {suspect.name}
No contradictions found in {suspect.name}'s statements.
They have been consistent so far. Keep questioning them!
"""
else:
contradictions_text = "\n\n".join([
f"**Contradiction {i+1}:**\n- Statement 1: \"{c.statement1}\"\n- Statement 2: \"{c.statement2}\"\n- Issue: {c.explanation}"
for i, c in enumerate(result.contradictions)
])
response = f"""# ⚠️ Contradictions Found: {suspect.name}
{contradictions_text}
These inconsistencies could indicate deception!
"""
return [TextContent(type="text", text=response)]
async def handle_get_timeline(args: dict) -> list[TextContent]:
"""Get the investigation timeline."""
session_id = args.get("session_id")
if not session_id:
return [TextContent(type="text", text="Error: session_id is required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
timeline = session.get_timeline()
if not timeline:
return [TextContent(type="text", text="No timeline events yet. Interrogate suspects to discover alibis and sightings.")]
response = "# 🕐 Investigation Timeline\n\n"
# Group by time slot
by_time = {}
for event in timeline:
time_slot = event.get("time_slot", "Unknown")
if time_slot not in by_time:
by_time[time_slot] = []
by_time[time_slot].append(event)
for time_slot in sorted(by_time.keys()):
response += f"## {time_slot}\n"
for event in by_time[time_slot]:
icon = {
"alibi_claim": "🗣️",
"witness_sighting": "👁️",
"clue_found": "🔎",
"contradiction": "⚠️"
}.get(event.get("event_type"), "📌")
suspect = event.get("suspect_name", "")
desc = event.get("description", "")
source = event.get("source", "")
response += f"- {icon} **{suspect}**: {desc}\n *(Source: {source})*\n"
response += "\n"
return [TextContent(type="text", text=response)]
# =============================================================================
# IMAGE GENERATION HANDLERS
# =============================================================================
async def handle_generate_scene_image(args: dict) -> list[TextContent]:
"""Generate an image of a location/scene."""
session_id = args.get("session_id")
location = args.get("location")
include_clue = args.get("include_clue", False)
if not all([session_id, location]):
return [TextContent(type="text", text="Error: session_id and location are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
from .image_generator import generate_scene
# Build the prompt
clue_context = ""
if include_clue:
# Find any clue at this location
for clue in session.discovered_clues:
if location.lower() in clue.lower():
clue_context = f" Include visual hint of: {clue}"
break
prompt = f"{location} in the style of a 1990s point-and-click adventure game.{clue_context} Setting: {session.setting}"
try:
result = await generate_scene(prompt, session.setting)
if result.error:
return [TextContent(type="text", text=f"Image generation failed: {result.error}")]
response = f"""# 🖼️ Scene: {location}
*Generated in {result.generation_time:.1f}s*
"""
return [TextContent(type="text", text=response)]
except Exception as e:
logger.exception("Scene generation error")
return [TextContent(type="text", text=f"Error generating scene: {str(e)}")]
async def handle_generate_portrait(args: dict) -> list[TextContent]:
"""Generate a portrait of a suspect."""
session_id = args.get("session_id")
suspect_name = args.get("suspect_name")
if not all([session_id, suspect_name]):
return [TextContent(type="text", text="Error: session_id and suspect_name are required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
# Find the suspect
suspect = session.find_suspect(suspect_name)
if not suspect:
available = ", ".join([s.name for s in session.public_suspects])
return [TextContent(type="text", text=f"Suspect '{suspect_name}' not found. Available: {available}")]
from .image_generator import generate_portrait
prompt = f"Character portrait of {suspect.name}, {suspect.role}. Personality: {suspect.personality}. Style: 1990s point-and-click adventure game character portrait."
try:
result = await generate_portrait(prompt, suspect.name)
if result.error:
return [TextContent(type="text", text=f"Portrait generation failed: {result.error}")]
response = f"""# 🎭 Portrait: {suspect.name}
**{suspect.role}**
*{suspect.personality}*
*Generated in {result.generation_time:.1f}s*
"""
return [TextContent(type="text", text=response)]
except Exception as e:
logger.exception("Portrait generation error")
return [TextContent(type="text", text=f"Error generating portrait: {str(e)}")]
async def handle_generate_title_card(args: dict) -> list[TextContent]:
"""Generate a title card for the mystery."""
session_id = args.get("session_id")
if not session_id:
return [TextContent(type="text", text="Error: session_id is required")]
session = get_session(session_id)
if not session.is_initialized:
return [TextContent(type="text", text="No game in progress. Use start_game first.")]
from .image_generator import generate_title_card
prompt = f"Title card for a murder mystery game. Setting: {session.setting}. Victim: {session.victim_name}. Style: 1990s LucasArts point-and-click adventure game title screen with dramatic lighting."
try:
result = await generate_title_card(prompt, session.victim_name)
if result.error:
return [TextContent(type="text", text=f"Title card generation failed: {result.error}")]
response = f"""# 🎬 Murder Mystery Title Card
**The Case of {session.victim_name}**
*{session.setting}*
*Generated in {result.generation_time:.1f}s*
"""
return [TextContent(type="text", text=response)]
except Exception as e:
logger.exception("Title card generation error")
return [TextContent(type="text", text=f"Error generating title card: {str(e)}")]
# =============================================================================
# RESOURCES
# =============================================================================
@server.list_resources()
async def list_resources() -> list[Resource]:
"""List available resources."""
resources = []
for session_id, session in sessions.items():
if session.is_initialized:
resources.extend([
Resource(
uri=f"mystery://{session_id}/state",
name=f"Game State ({session_id[:8]}...)",
description="Current game state including suspects, clues, and progress",
mimeType="application/json"
),
Resource(
uri=f"mystery://{session_id}/suspects",
name=f"Suspects ({session_id[:8]}...)",
description="List of suspects with their public information",
mimeType="application/json"
),
Resource(
uri=f"mystery://{session_id}/clues",
name=f"Discovered Clues ({session_id[:8]}...)",
description="All clues discovered during investigation",
mimeType="application/json"
),
])
return resources
@server.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
"""List resource templates."""
return [
ResourceTemplate(
uriTemplate="mystery://{session_id}/state",
name="Game State",
description="Get the current state of a game session"
),
ResourceTemplate(
uriTemplate="mystery://{session_id}/suspects",
name="Suspects List",
description="Get the list of suspects"
),
ResourceTemplate(
uriTemplate="mystery://{session_id}/clues",
name="Discovered Clues",
description="Get all discovered clues"
),
]
@server.read_resource()
async def read_resource(uri: str) -> str:
"""Read a resource."""
# Parse URI: mystery://{session_id}/{resource_type}
parts = uri.replace("mystery://", "").split("/")
if len(parts) < 2:
return json.dumps({"error": "Invalid URI format"})
session_id, resource_type = parts[0], parts[1]
session = sessions.get(session_id)
if not session or not session.is_initialized:
return json.dumps({"error": "Session not found or not initialized"})
if resource_type == "state":
return json.dumps({
"session_id": session_id,
"setting": session.setting,
"victim": {
"name": session.victim_name,
"background": session.victim_background
},
"suspects_talked_to": list(session.suspects_talked_to),
"discovered_clues": session.discovered_clues,
"searched_locations": list(session.searched_locations),
"available_locations": session.available_locations,
"wrong_accusations": session.wrong_accusations,
"game_over": session.game_over,
"won": session.won,
"turn": session.turn
}, indent=2)
elif resource_type == "suspects":
return json.dumps([
{
"name": s.name,
"role": s.role,
"personality": s.personality,
"alibi": s.alibi if s.name in session.suspects_talked_to else "(Not yet revealed - interrogate this suspect)",
"interrogated": s.name in session.suspects_talked_to,
"trust": session.get_suspect_trust(s.name),
"nervousness": session.get_suspect_nervousness(s.name)
}
for s in session.public_suspects
], indent=2)
elif resource_type == "clues":
return json.dumps({
"discovered": session.discovered_clues,
"count": len(session.discovered_clues)
}, indent=2)
else:
return json.dumps({"error": f"Unknown resource type: {resource_type}"})
# =============================================================================
# MAIN
# =============================================================================
async def main():
"""Run the MCP server."""
logger.info("Starting Murder Mystery MCP Server...")
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
