一个用于顺序思考和问题解决的 Python MCP 服务器
</div>增强的 Python 版本:由 Anthropic 开发的 顺序思考 MCP 服务器 的 Python 端口。在保持完全兼容性的同时,增加了置信评分、自动分配的思考编号和多会话支持。
[!NOTE] 元信息:此 MCP 服务器是使用 UltraThink 迭代构建的——这是该工具能力的一个实际示例,能够分解复杂问题、管理架构决策并在开发会话之间保持上下文。
直接从 GitHub 使用 uvx 运行(无需安装):
uvx --from git+https://github.com/husniadil/ultrathink ultrathink
本地开发:
# 克隆仓库
git clone https://github.com/husniadil/ultrathink.git
cd ultrathink
# 安装所有依赖项(包括开发依赖项)
uv sync
# 列出所有可用的任务
uv run task --list
# 启动服务器
uv run task run
# 带覆盖率运行测试
uv run task test
# 不带覆盖率快速运行测试
uv run task test-quick
# 运行测试客户端
uv run task client
# 格式化代码(ruff + prettier)
uv run task format
# 代码检查
uv run task lint
# 使用 mypy 进行类型检查
uv run task typecheck
# 清理缓存文件
uv run task clean
不使用任务运行器直接执行:
# 直接运行服务器
uv run ultrathink
# 直接运行测试客户端
uv run python examples/client.py
注意:对于测试、检查和格式化,请优先使用上面显示的 uv run task 命令。
服务器提供了一个单一工具,通过结构化的思考进行动态和反思的问题解决。
必需的:
thought (str):当前思考步骤total_thoughts (int):预计所需的总思考数(>=1)可选的:
thought_number (int):当前思考编号 - 如果省略,则按顺序自动分配(1, 2, 3...),或提供显式的编号以控制分支/语义next_thought_needed (bool):是否需要另一个思考步骤。如果省略,则自动分配为 thought_number < total_thoughts。显式设置以覆盖默认行为session_id (str):会话标识符,用于管理多个思考会话(None = 创建新的,提供ID以继续会话)is_revision (bool):这是否修订了之前的思考revises_thought (int):正在重新考虑的思考编号branch_from_thought (int):分支点思考编号branch_id (str):分支标识符needs_more_thoughts (bool):是否需要更多的思考confidence (float):置信水平(0.0-1.0,例如,0.7 表示70%置信)uncertainty_notes (str):关于这个思考的疑虑或关注的可选解释outcome (str):这个思考所实现或预期的结果assumptions (list[Assumption]):在这个思考中做出的假设(id, text, confidence, critical, verifiable)depends_on_assumptions (list[str]):这个思考依赖的假设ID(例如,["A1", "A2"])invalidates_assumptions (list[str]):证明为假的假设ID(例如,["A3"])返回一个 JSON 对象,包含:
session_id:会话标识符,用于继续会话thought_number:当前思考编号total_thoughts:总思考数(如果需要,自动调整)next_thought_needed:是否需要更多的思考branches:分支ID列表thought_history_length:在此会话中处理的思考数量confidence:此思考的置信水平(0.0-1.0,可选)uncertainty_notes:疑虑或关注的解释(可选)outcome:实现或预期的结果(可选)all_assumptions:在此会话中跟踪的所有假设(按ID键)risky_assumptions:风险假设的ID(关键且置信度低且未验证)falsified_assumptions:被证明为假的假设IDfrom fastmcp import Client
from ultrathink import mcp
async with Client(mcp) as client:
# 简单的顺序思考,自动分配字段
result = await client.call_tool("ultrathink", {
"thought": "让我逐步分析这个问题",
"total_thoughts": 3
# thought_number 自动分配:1
# next_thought_needed 自动分配:True (1 < 3)
})
async with Client(mcp) as client:
# 带置信评分和显式会话
result = await client.call_tool("ultrathink", {
"thought": "初步假设 - 这种方法可能有效",
"total_thoughts": 5,
"confidence": 0.6, # 60%置信
# next_thought_needed 自动分配:True
"session_id": "problem-solving-session-1"
})
# 继续相同的会话,置信度更高
result2 = await client.call_tool("ultrathink", {
"thought": "经过分析,我对这个解决方案更有信心",
"total_thoughts": 5,
"confidence": 0.9, # 90%置信
# next_thought_needed 自动分配:True
"session_id": "problem-solving-session-1" # 相同会话
})
# 从之前的思考分支
result3 = await client.call_tool("ultrathink", {
"thought": "让我探索另一种方法",
"total_thoughts": 6,
"confidence": 0.7,
"branch_from_thought": 1,
"branch_id": "alternative-path",
# next_thought_needed 自动分配:True
"session_id": "problem-solving-session-1"
})
async with Client(mcp) as client:
# 跟踪不确定性和结果
result = await client.call_tool("ultrathink", {
"thought": "测试身份验证修复",
"total_thoughts": 5,
"confidence": 0.8,
"uncertainty_notes": "尚未在高负载下测试",
"outcome": "登录流程对标准用户有效"
})
# 响应包括新字段
print(result["confidence"]) # 0.8
print(result["uncertainty_notes"]) # "尚未在高负载下测试"
print(result["outcome"]) # "登录流程对标准用户有效"
async with Client(mcp) as client:
# 思考1:明确声明假设
result = await client.call_tool("ultrathink", {
"thought": "Redis 应该满足我们的性能要求",
"total_thoughts": 4,
"assumptions": [
{
"id": "A1",
"text": "到 Redis 的网络延迟 < 5ms",
"confidence": 0.8,
"critical": True,
"verifiable": True,
"evidence": "基于预演环境中的初步网络测试"
}
]
})
# 思考2:基于之前的假设
result2 = await client.call_tool("ultrathink", {
"thought": "基于低延迟,Redis 可以处理 10K req/sec",
"total_thoughts": 4,
"depends_on_assumptions": ["A1"],
"session_id": result["session_id"]
})
# 思考3:如果证明为假则无效
result3 = await client.call_tool("ultrathink", {
"thought": "经过测试,延迟是 15ms,不是 5ms!",
"total_thoughts": 4,
"invalidates_assumptions": ["A1"],
"session_id": result["session_id"]
})
# 跟踪所有假设并检测风险假设
print(result3["all_assumptions"]) # {"A1": {...}}
print(result3["falsified_assumptions"]) # ["A1"]
DISABLE_THOUGHT_LOGGING:设置为 "true" 以禁用将思考日志输出到 stderr添加到你的 claude_desktop_config.json:
{
"mcpServers": {
"UltraThink": {
"command": "uvx",
"args": [
"--from",
"git+https://github.com/husniadil/ultrathink",
"ultrathink"
]
}
}
}
从源代码进行本地开发:
{
"mcpServers": {
"UltraThink": {
"command": "uv",
"args": ["--directory", "/path/to/ultrathink", "run", "ultrathink"]
}
}
}
为了本地开发和测试,你可以创建一个 .mcp.json 文件(参见 .mcp.json.example):
# 复制示例文件
cp .mcp.json.example .mcp.json
# 编辑以匹配你的本地路径
# 将 /path/to/ultrathink 更改为你的实际目录
示例配置(.mcp.json.example):
{
"mcpServers": {
"UltraThink": {
"command": "uv",
"args": ["--directory", "/path/to/ultrathink", "run", "ultrathink"],
"env": {
"DISABLE_THOUGHT_LOGGING": "false"
}
}
}
}
此配置:
DISABLE_THOUGHT_LOGGING: "false").mcp.json 配置的 MCP 客户端.mcp.json 被 git 忽略 - 为你的本地设置自定义它重要:会话仅存储在内存中,并且当服务器重启或终止时会丢失。每个会话由一个唯一的会话ID标识,并维护:
影响:
最佳实践:
根据简单分层架构原则构建,以实现职责的清晰分离和可维护的代码。
src/ultrathink/(三层结构)
模型层(models/)
服务层(services/)
接口层(interface/)
包入口点
uv run ultrathink)tests/(100% 覆盖,镜像源结构)
模型测试(models/)
服务测试(services/)
接口测试(interface/)
根测试文件
examples/
用于数据表示和验证的 Pydantic 模型:
思考:核心模型,代表一个带有验证和行为的单个思考 思考请求:来自 MCP 客户端的输入模型,带有验证 思考响应:输出模型,向 MCP 客户端提供结构化数据 思考会话:管理思考历史和分支的会话模型
# 类型安全模型使用
request = ThoughtRequest(
thought="我的思考步骤",
thought_number=1,
total_thoughts=3,
next_thought_needed=True
)
response = ThoughtResponse(
thought_number=1,
total_thoughts=3,
next_thought_needed=True,
branches=[],
thought_history_length=1
)
业务逻辑和编排:
UltraThinkService:编排思考过程
职责:
思考请求 → 思考 模型(输入)思考会话思考响应 模型(输出)关键方法:
process_thought(request: 思考请求) → 思考响应:主要编排service = UltraThinkService()
# 完整流程:
# 1. 接收来自接口层的思考请求
# 2. 转换为思考模型
# 3. 调用 session.add_thought()(业务逻辑)
# 4. 从会话状态构建思考响应
# 5. 返回响应
request = 思考请求(thought="...", thought_number=1, ...)
response = service.process_thought(request)
使用 FastMCP 的外部接口:
mcp_server.py:MCP 服务器工具注册
职责:
@mcp.tool 装饰器定义 MCP 工具@mcp.tool
def ultrathink(thought: str, total_thoughts: int, ...) -> 思考响应:
request = 思考请求(thought=thought, total_thoughts=total_thoughts, ...)
return thinking_service.process_thought(request)
类型安全优势:
职责清晰分离:
更简单的结构:扁平的文件夹层次结构(2级而不是3级)
更容易导入:较短的相对导入路径(..models 而