首页让大模型当地下城主持人，讲故事的部分它做得很好；但让它同时管生命值、掷骰和掉落，就会出现「刚才你还剩 30 点血，现在怎么变成 50 了」这种问题。同一个动作问两遍，可能得到两个互相矛盾的结果。
这篇教程的分工很明确：DeepSeek 只负责中文叙事，角色、战斗、随机数全交给确定性的 MCP 工具，进度存 SQLite。模型讲故事，引擎管数值，各干各的。
玩起来你会发现：每一个数字都能在工具返回值里找到出处；重启 MCP 之后，存档还在。
> 开始之前 游戏工具走 MCP，先看懂《给语音智能体接入 MCP 工具：让它真的去查订单》的接入方式。
一. 架构与准备工作
玩家语音 → 凤鸣 → DeepSeek 地下城主持人
│ MCP tool call
▼
Streamable HTTP MCP
│
SQLite + 确定性游戏规则
│ 工具结果
▼
DeepSeek 中文叙事 → MiniMax 播报需要 App ID、主要证书、DeepSeek、MiniMax，以及云端能访问的 HTTPS MCP 地址。依赖固定为 agora-agents==2.4.1 和 mcp==1.12.3。
交付代码是单人演示，一个 MCP 进程对应一个存档。真要多人上线，必须按已认证用户或房间隔离数据库——所有玩家共用一个 rpg.db，会变成大家在同一具身体里冒险。
二. 环境变量
AGORA_APP_ID=REPLACE_WITH_SHENGWANG_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_PRIMARY_CERTIFICATE
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_KEY
MCP_ENDPOINT=https://REPLACE_WITH_PUBLIC_HOST/mcp
MCP_SHARED_SECRET=REPLACE_WITH_RANDOM_SHARED_SECRET
RPG_DB_PATH=rpg.db三. 确定性游戏引擎
import json
import random
import sqlite3
CLASSES = {
"战士": {"hp": 32, "die": 8, "spell": "盾击"},
"法师": {"hp": 20, "die": 6, "spell": "火球术"},
"游侠": {"hp": 24, "die": 6, "spell": "连射"},
"祭司": {"hp": 26, "die": 6, "spell": "圣光"},
}
def connect(path: str = "rpg.db") -> sqlite3.Connection:
db = sqlite3.connect(path)
db.executescript("""
CREATE TABLE IF NOT EXISTS character (
id INTEGER PRIMARY KEY CHECK(id=1),
class_name TEXT, hp INTEGER, max_hp INTEGER,
gold INTEGER, inventory TEXT
);
CREATE TABLE IF NOT EXISTS enemy (
id INTEGER PRIMARY KEY CHECK(id=1),
name TEXT, hp INTEGER, attack INTEGER
);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY, value TEXT
);
INSERT OR IGNORE INTO settings(key, value)
VALUES ('mode', 'narration');
""")
db.commit()
return db
def create_character(
db: sqlite3.Connection,
character_class: str,
) -> str:
if character_class not in CLASSES:
return "请选择职业：战士、法师、游侠或祭司。"
stats = CLASSES[character_class]
db.execute("DELETE FROM character")
db.execute("DELETE FROM enemy")
db.execute(
"""INSERT INTO character(
id, class_name, hp, max_hp, gold, inventory
) VALUES (1, ?, ?, ?, 0, ?)""",
(
character_class,
stats["hp"],
stats["hp"],
json.dumps([], ensure_ascii=False),
),
)
db.execute(
"UPDATE settings SET value='narration' WHERE key='mode'"
)
db.commit()
return (
f"你成为了{character_class}，生命值 {stats['hp']}，"
f"招牌技能是{stats['spell']}。"
)
def roll_damage(sides: int, rng: random.Random) -> int:
return rng.randint(1, sides)伤害、反击、掉落、状态更新全部在 game.py 里算完。LLM 只能复述工具返回的结果，不允许自己生成掷骰点数——这是整套设计的地基。
四. 暴露 MCP 工具
from mcp.server.fastmcp import FastMCP
import game
mcp = FastMCP("shengwang-voice-rpg")
def run_game(function, *arguments) -> str:
db = game.connect()
try:
return function(db, *arguments)
finally:
db.close()
@mcp.tool()
def create_character(character_class: str) -> str:
"""创建角色；职业必须是战士、法师、游侠或祭司。"""
return run_game(game.create_character, character_class)
@mcp.tool()
def get_character() -> str:
"""查询职业、生命、金币、物品和当前战斗状态。"""
return run_game(game.get_character)
@mcp.tool()
def start_encounter() -> str:
"""开始随机遭遇。"""
return run_game(game.start_encounter)
@mcp.tool()
def attack() -> str:
"""完成普通攻击、反击和战利品结算。"""
return run_game(game.attack)
@mcp.tool()
def cast_spell(spell_name: str) -> str:
"""施放角色技能并结算本轮战斗。"""
return run_game(game.cast_spell, spell_name)
@mcp.tool()
def flee() -> str:
"""逃离当前战斗。"""
return run_game(game.flee)完整服务还在 ASGI 外层强制校验 Authorization: Bearer，并且只挂载 Streamable HTTP MCP。
五. 配置智能体
import os
from agora_agent import Area, AsyncAgora
from agora_agent.agentkit import Agent
from agora_agent.cn import DeepSeekLLM, FengmingSTT, MiniMaxTTS
client = AsyncAgora(
area=Area.CN,
app_id=os.environ["AGORA_APP_ID"],
app_certificate=os.environ["AGORA_APP_CERTIFICATE"],
)
mcp_servers = [{
"name": "voice-rpg",
"endpoint": os.environ["MCP_ENDPOINT"],
"transport": "streamable_http",
"headers": {
"Authorization": f"Bearer {os.environ['MCP_SHARED_SECRET']}"
},
"allowed_tools": [
"create_character", "get_character", "start_encounter",
"attack", "cast_spell", "flee",
],
"timeout_ms": 10000,
}]
llm = DeepSeekLLM(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url="https://api.deepseek.com/chat/completions",
model="deepseek-v4-flash",
system_messages=[{"role": "system", "content":
"你是中文奇幻语音 RPG 的地下城主持人。所有职业、生命、"
"掷骰、伤害、金币、物品和战斗结果必须调用工具，绝不能编造。"
"工具返回后只能基于工具结果继续叙述。"}],
greeting_message=(
"欢迎来到雾岭大陆。请先选择战士、法师、游侠或祭司。"
),
max_history=30,
max_tokens=512,
temperature=0.6,
params={"thinking": {"type": "disabled"}},
mcp_servers=mcp_servers,
)
tts = MiniMaxTTS(
key=os.environ["MINIMAX_API_KEY"],
model="speech-01-turbo",
voice_id="female-shaonv",
sample_rate=16000,
language_boost="Chinese",
)
agent = (
Agent(
client=client,
turn_detection={"language": "zh-CN"},
advanced_features={"enable_rtm": True, "enable_tools": True},
parameters={"audio_scenario": "chorus", "data_channel": "rtm"},
)
.with_stt(FengmingSTT())
.with_llm(llm)
.with_tts(tts)
)六. 运行与验证
cd code/voice-rpg
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn mcp_service:app --host 0.0.0.0 --port 9500把 9500 端口部署成 MCP_ENDPOINT 对应的公网 HTTPS，再启动智能体后端：
uvicorn app:app --host 0.0.0.0 --port 8000按这个脚本玩一遍：
玩家：我要当法师。 → create_character
玩家：查看我的状态。 → get_character
玩家：我要探索洞穴。 → start_encounter
玩家：施放火球术。 → cast_spell
玩家：继续攻击。 → attack验收标准很硬：每一个数字都能在 SQLite 的工具结果里找到出处，重启 MCP 后存档还在。
还要试试这些边界情况：没建角色就开打、重复开始战斗、用不存在的技能、角色死亡、逃跑、MCP 返回 401、工具超时，以及最关键的一条——确认 DeepSeek 没法凭空修改生命值。
七. 故障排查
- LLM 编造伤害数字：系统提示里要强制要求调用工具，同时检查
enable_tools、mcp_servers和allowed_tools配置正确。 - MCP 返回 401：智能体配置里的 Bearer 要和服务端
MCP_SHARED_SECRET一致。 - 所有玩家共享存档：教程默认单进程单玩家。生产环境按认证用户隔离存储，或者为每局部署独立的游戏实例。
- 最后再强调一次：随机数、结算、状态变更都要留在可测试的确定性服务端。交给 LLM 的那一刻，游戏就没有规则了。
八. 下一步
想给游戏加更多工具，先回到 MCP 那篇把工具接入这件事吃透。
《给语音智能体接入 MCP 工具：让它真的去查订单》
