跳到主要内容

第五部分：MCP 进阶示例

如前文所述，MCP包括MCP Server 服务端和MCP Client客户端两个部分，服务端为客户端提供相应的服务，本部分作为进阶训练，我们来撰写一个证券账户的MCP Client模拟Agent与外界交互的过程

5.1 完整 MCP 示例：证券持仓管理系统​

让我们看一个完整的 MCP 服务器示例，理解 Tool、Resource、Prompt 如何配合使用。

文件：src/cufel_deepagent/mcp/servers/stock_server.py

数据初始化与数据持久化​

点击展开：完整代码（共39行）
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("证券持仓管理系统")
DATA_FILE = Path(__file__).parent / "stock_portfolio.json"

def load_portfolio():
    """从 JSON 文件加载持仓数据"""
    if not DATA_FILE.exists():
        initial_data = {
            "portfolio_id": "PORTFOLIO-001",
            "account_name": "主账户",
            "positions": [
                {"name": "深空智能", "symbol": "688588", "quantity": 500,
                 "avg_cost": 85.0, "current_price": 72.5,
                 "take_profit": 102.0, "stop_loss": 76.5},
                {"name": "虚研科技", "symbol": "300789", "quantity": 300,
                 "avg_cost": 45.0, "current_price": 58.5,
                 "take_profit": 54.0, "stop_loss": 36.0}
            ]
        }
        save_portfolio(initial_data)
        return initial_data
    with open(DATA_FILE, "r", encoding="utf-8") as f:
        return json.load(f)

def save_portfolio(data):
    """保存持仓数据到 JSON 文件"""
    with open(DATA_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=4, ensure_ascii=False)

def find_position(data, name):
    """根据名称查找持仓"""
    for pos in data["positions"]:
        if pos["name"] == name:
            return pos
    return None

作用：定义 MCP 服务器、初始化数据存储文件、封装数据读写和查询函数。

Resource（Agent 的眼睛）​

@mcp.resource("portfolio://overview")
def get_portfolio_overview() -> str:
    """查看当前证券持仓状态"""
    pf = load_portfolio()
    lines = [f"--- 证券持仓 [{pf['portfolio_id']}] ---", "当前持仓:"]
    for pos in pf["positions"]:
        change = ((pos["current_price"] - pos["avg_cost"]) / pos["avg_cost"]) * 100
        status = "🔴 需清仓" if pos["current_price"] >= pos["take_profit"] or pos["current_price"] <= pos["stop_loss"] else "🟢 正常"
        lines.append(f"  {pos['name']}({pos['symbol']}): {pos['current_price']}元 | 盈亏:{change:+.1f}% | {status}")
    return "\n".join(lines)

作用：Agent 可以随时读取持仓状态，了解当前资产情况。

Tool（Agent 的手）​

点击展开：完整代码（共17行）
@mcp.tool()
def update_price(item_name: str, new_price: float) -> str:
    """更新股票价格并检查止盈止损"""
    pos = find_position(load_portfolio(), item_name)
    if not pos:
        return f"股票不存在"

    pos["current_price"] = new_price
    save_portfolio(load_portfolio())

    # 检查是否触发止盈止损
    if new_price >= pos["take_profit"]:
        return f"🔴 【止盈触发】{item_name} 当前价 {new_price}元 >= 止盈价 {pos['take_profit']}元，建议清仓！"
    elif new_price <= pos["stop_loss"]:
        return f"🔴 【止损触发】{item_name} 当前价 {new_price}元 <= 止损价 {pos['stop_loss']}元，建议清仓！"
    return f"🟢 {item_name} 当前价 {new_price}元，在正常范围内"

作用：Agent 可以执行股票交易操作。

Prompt（Agent 的指南）​

点击展开：完整代码（共17行）
@mcp.prompt()
def stock_strategy() -> str:
    """证券投资策略指导"""
    return """
    你是证券投资专家。请按以下规则操作：

    1. 使用 portfolio://overview 查看持仓
    2. 价格更新后调用 update_price 检查止盈止损
    3. 止盈条件：当前价 >= 止盈价 → 清仓锁定收益
    4. 止损条件：当前价 <= 止损价 → 清仓控制风险
    5. 正常持有：止损价 < 当前价 < 止盈价 → 不操作

    示例：股票成本100元，止盈120元，止损80元
    - 涨到125元 → 触发止盈 → 清仓
    - 跌到75元 → 触发止损 → 清仓
    """

作用：告诉 Agent 遇到证券投资问题该如何思考和决策。

5.2 MCP 三大组件总结​

组件问题答案装饰器
ResourceAgent 想知道什么？账户状态、余额@mcp.resource()
ToolAgent 能做什么？买入、卖出、查询@mcp.tool()
PromptAgent 该怎么做？投资策略、决策逻辑@mcp.prompt()