44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
"""测试 SSE MCP 客户端连接"""
|
|
import asyncio
|
|
import sys
|
|
|
|
from mcp.client.sse import sse_client
|
|
from mcp.client.session import ClientSession
|
|
|
|
|
|
async def main():
|
|
url = "http://127.0.0.1:8001/mcp/sse"
|
|
print(f"连接到: {url}")
|
|
|
|
try:
|
|
async with sse_client(url) as (read_stream, write_stream):
|
|
print("SSE 连接建立成功！")
|
|
|
|
async with ClientSession(read_stream, write_stream) as session:
|
|
print("正在初始化 MCP 会话...")
|
|
result = await session.initialize()
|
|
print(f"初始化成功! Server: {result.serverInfo.name} v{result.serverInfo.version}")
|
|
|
|
print("\n列出工具...")
|
|
tools = await session.list_tools()
|
|
print(f"工具数量: {len(tools.tools)}")
|
|
for t in tools.tools:
|
|
print(f" - {t.name}: {t.description[:60]}")
|
|
|
|
print("\n提交一个测试任务...")
|
|
result = await session.call_tool("submit_task", {
|
|
"description": "这是一个测试任务，请回复 hello world",
|
|
"project_path": "D:\\nzy\\workspace_git",
|
|
})
|
|
print(f"结果: {result}")
|
|
|
|
except Exception as e:
|
|
print(f"连接失败: {type(e).__name__}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|