249 lines
7.0 KiB
Python
249 lines
7.0 KiB
Python
"""
|
||
MCP客户端使用示例
|
||
演示如何调用MCP服务器的各种工具
|
||
"""
|
||
|
||
import httpx
|
||
import json
|
||
from typing import Dict, Any
|
||
|
||
|
||
class MCPClient:
|
||
"""MCP客户端"""
|
||
|
||
def __init__(self, base_url: str = "http://localhost:8900"):
|
||
self.base_url = base_url
|
||
self.client = httpx.Client(timeout=60.0)
|
||
|
||
def list_tools(self):
|
||
"""列出所有可用工具"""
|
||
response = self.client.get(f"{self.base_url}/tools")
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
def get_tool(self, tool_name: str):
|
||
"""获取特定工具的定义"""
|
||
response = self.client.get(f"{self.base_url}/tools/{tool_name}")
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
def call_tool(self, tool_name: str, arguments: Dict[str, Any]):
|
||
"""调用工具"""
|
||
payload = {
|
||
"tool": tool_name,
|
||
"arguments": arguments
|
||
}
|
||
response = self.client.post(f"{self.base_url}/tools/call", json=payload)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
def close(self):
|
||
"""关闭客户端"""
|
||
self.client.close()
|
||
|
||
|
||
def print_result(title: str, result: Dict[str, Any]):
|
||
"""打印结果"""
|
||
print(f"\n{'=' * 60}")
|
||
print(f"{title}")
|
||
print(f"{'=' * 60}")
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
|
||
|
||
def main():
|
||
"""主函数 - 演示各种工具的使用"""
|
||
|
||
client = MCPClient()
|
||
|
||
try:
|
||
# 1. 列出所有工具
|
||
print("\n示例1: 列出所有可用工具")
|
||
tools = client.list_tools()
|
||
print(f"可用工具数量: {len(tools['tools'])}")
|
||
for tool in tools['tools']:
|
||
print(f" - {tool['name']}: {tool['description'][:50]}...")
|
||
|
||
# 2. 搜索中国新闻
|
||
print("\n示例2: 搜索中国新闻（关键词：人工智能）")
|
||
result = client.call_tool(
|
||
"search_china_news",
|
||
{
|
||
"query": "人工智能",
|
||
"top_k": 5
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("中国新闻搜索结果", result['data'])
|
||
|
||
# 3. 搜索概念板块（按涨跌幅排序）
|
||
print("\n示例3: 搜索概念板块（关键词：新能源，按涨跌幅排序）")
|
||
result = client.call_tool(
|
||
"search_concepts",
|
||
{
|
||
"query": "新能源",
|
||
"size": 5,
|
||
"sort_by": "change_pct"
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("概念搜索结果", result['data'])
|
||
|
||
# 4. 获取股票的相关概念
|
||
print("\n示例4: 获取股票相关概念（股票代码：600519）")
|
||
result = client.call_tool(
|
||
"get_stock_concepts",
|
||
{
|
||
"stock_code": "600519",
|
||
"size": 10
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("股票概念结果", result['data'])
|
||
|
||
# 5. 搜索涨停股票
|
||
print("\n示例5: 搜索涨停股票（关键词：锂电池）")
|
||
result = client.call_tool(
|
||
"search_limit_up_stocks",
|
||
{
|
||
"query": "锂电池",
|
||
"mode": "hybrid",
|
||
"page_size": 5
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("涨停股票搜索结果", result['data'])
|
||
|
||
# 6. 搜索研究报告
|
||
print("\n示例6: 搜索研究报告（关键词：投资策略）")
|
||
result = client.call_tool(
|
||
"search_research_reports",
|
||
{
|
||
"query": "投资策略",
|
||
"mode": "hybrid",
|
||
"size": 3
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("研究报告搜索结果", result['data'])
|
||
|
||
# 7. 获取概念统计数据
|
||
print("\n示例7: 获取概念统计（最近7天）")
|
||
result = client.call_tool(
|
||
"get_concept_statistics",
|
||
{
|
||
"days": 7,
|
||
"min_stock_count": 3
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("概念统计结果", result['data'])
|
||
|
||
# 8. 搜索路演信息
|
||
print("\n示例8: 搜索路演信息（关键词：业绩）")
|
||
result = client.call_tool(
|
||
"search_roadshows",
|
||
{
|
||
"query": "业绩",
|
||
"size": 3
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("路演搜索结果", result['data'])
|
||
|
||
# 9. 获取股票基本信息
|
||
print("\n示例9: 获取股票基本信息（股票：600519）")
|
||
result = client.call_tool(
|
||
"get_stock_basic_info",
|
||
{
|
||
"seccode": "600519"
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("股票基本信息", result['data'])
|
||
|
||
# 10. 获取股票财务指标
|
||
print("\n示例10: 获取股票财务指标（股票：600519，最近5期）")
|
||
result = client.call_tool(
|
||
"get_stock_financial_index",
|
||
{
|
||
"seccode": "600519",
|
||
"limit": 5
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("财务指标", result['data'])
|
||
|
||
# 11. 获取股票交易数据
|
||
print("\n示例11: 获取股票交易数据（股票：600519，最近10天）")
|
||
result = client.call_tool(
|
||
"get_stock_trade_data",
|
||
{
|
||
"seccode": "600519",
|
||
"limit": 10
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("交易数据", result['data'])
|
||
|
||
# 12. 按行业搜索股票
|
||
print("\n示例12: 按行业搜索股票（行业：半导体）")
|
||
result = client.call_tool(
|
||
"search_stocks_by_criteria",
|
||
{
|
||
"industry": "半导体",
|
||
"limit": 10
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("行业股票", result['data'])
|
||
|
||
# 13. 股票对比分析
|
||
print("\n示例13: 股票对比分析（600519 vs 000858）")
|
||
result = client.call_tool(
|
||
"get_stock_comparison",
|
||
{
|
||
"seccodes": ["600519", "000858"],
|
||
"metric": "financial"
|
||
}
|
||
)
|
||
if result['success']:
|
||
print_result("股票对比", result['data'])
|
||
|
||
except Exception as e:
|
||
print(f"\n错误: {str(e)}")
|
||
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def test_single_tool():
|
||
"""测试单个工具（用于快速测试）"""
|
||
client = MCPClient()
|
||
|
||
try:
|
||
# 修改这里来测试不同的工具
|
||
result = client.call_tool(
|
||
"search_china_news",
|
||
{
|
||
"query": "芯片",
|
||
"exact_match": True,
|
||
"top_k": 3
|
||
}
|
||
)
|
||
|
||
print_result("测试结果", result)
|
||
|
||
except Exception as e:
|
||
print(f"错误: {str(e)}")
|
||
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 运行完整示例
|
||
main()
|
||
|
||
# 或者测试单个工具
|
||
# test_single_tool()
|