demo.py•3.06 kB
#!/usr/bin/env python3
"""
MCP服务器演示脚本
展示如何与员工管理MCP服务器交互
"""
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def demo_mcp_server():
"""演示MCP服务器功能"""
print("🚀 开始演示MCP服务器功能")
print("=" * 50)
# 连接到MCP服务器
server_params = StdioServerParameters(
command="python",
args=["src/server.py"]
)
try:
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
print("✅ 已连接到MCP服务器")
# 1. 获取可用工具列表
print("\n📋 获取可用工具列表:")
tools = await session.list_tools()
for i, tool in enumerate(tools.tools, 1):
print(f" {i}. {tool.name}: {tool.description}")
# 2. 演示工具调用
print("\n🔧 演示工具调用:")
# 获取所有员工
print("\n1. 获取所有员工:")
result = await session.call_tool("get_all_employees", {})
if result.content:
for content in result.content:
print(f" {content.text}")
# 添加新员工
print("\n2. 添加新员工:")
add_result = await session.call_tool("add_employee", {
"firstName": "张三",
"lastName": "李四",
"salary": 8000.0,
"currency": "CNY",
"birthdate": "1990-01-01",
"isActive": True,
"level": "高级"
})
if add_result.content:
for content in add_result.content:
print(f" {content.text}")
# 搜索员工
print("\n3. 搜索员工:")
search_result = await session.call_tool("search_employees", {
"lastName": "李"
})
if search_result.content:
for content in search_result.content:
print(f" {content.text}")
# 获取在职员工
print("\n4. 获取在职员工:")
active_result = await session.call_tool("get_active_employees", {})
if active_result.content:
for content in active_result.content:
print(f" {content.text}")
print("\n✅ 演示完成！")
except Exception as e:
print(f"❌ 演示过程中出现错误: {e}")
if __name__ == "__main__":
asyncio.run(demo_mcp_server())