import unittest
import asyncio
import aiohttp
import json
import os
class TestMcpSequentialThinkingTqsTools(unittest.TestCase):
"""
模拟调用mcp-sequential-thinking-tqs的所有工具，
并打印每次请求的入参和返回结果。
"""
SERVER_URL = os.environ.get("MCP_SERVER_URL", "http://127.0.0.1:8000")
MESSAGE_URL = f"{SERVER_URL}/messages/"
SESSION_ID = "test-session-001"
def test_all_tools(self):
asyncio.run(self.async_test_all_tools())
async def async_test_all_tools(self):
async with aiohttp.ClientSession() as session:
# 1. process_thought
process_thought_args = {
"thought": "测试思考内容",
"thought_number": 1,
"total_thoughts": 1,
"next_thought_needed": False,
"stage": "Problem Definition",
"tags": ["单元测试", "工具调用"],
"axioms_used": ["测试公理"],
"assumptions_challenged": ["假设1"]
}
process_thought_req = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "process_thought",
"arguments": process_thought_args,
"session_id": self.SESSION_ID
}
}
print("\n【process_thought】请求参数:")
print(json.dumps(process_thought_req, ensure_ascii=False, indent=2))
async with session.post(self.MESSAGE_URL, json=process_thought_req) as resp:
result = await resp.json()
print("【process_thought】返回结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 2. generate_summary
generate_summary_req = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "generate_summary",
"arguments": {},
"session_id": self.SESSION_ID
}
}
print("\n【generate_summary】请求参数:")
print(json.dumps(generate_summary_req, ensure_ascii=False, indent=2))
async with session.post(self.MESSAGE_URL, json=generate_summary_req) as resp:
result = await resp.json()
print("【generate_summary】返回结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 3. clear_history
clear_history_req = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "clear_history",
"arguments": {},
"session_id": self.SESSION_ID
}
}
print("\n【clear_history】请求参数:")
print(json.dumps(clear_history_req, ensure_ascii=False, indent=2))
async with session.post(self.MESSAGE_URL, json=clear_history_req) as resp:
result = await resp.json()
print("【clear_history】返回结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 4. export_session
export_path = "./test_export_session.json"
export_session_req = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "export_session",
"arguments": {"file_path": export_path},
"session_id": self.SESSION_ID
}
}
print("\n【export_session】请求参数:")
print(json.dumps(export_session_req, ensure_ascii=False, indent=2))
async with session.post(self.MESSAGE_URL, json=export_session_req) as resp:
result = await resp.json()
print("【export_session】返回结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 5. import_session
import_session_req = {
"jsonrpc": "2.0",
"id": 5,
"method": "tools/call",
"params": {
"name": "import_session",
"arguments": {"file_path": export_path},
"session_id": self.SESSION_ID
}
}
print("\n【import_session】请求参数:")
print(json.dumps(import_session_req, ensure_ascii=False, indent=2))
async with session.post(self.MESSAGE_URL, json=import_session_req) as resp:
result = await resp.json()
print("【import_session】返回结果:")
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
unittest.main()
