前言:

  整个过程,回头来看确实是很简单的,但不知道为什么豆包之类的AI终端,甚至是AI编程模块,给的都是假代码,要不然就是乱引用不存在的库,要不然就是简单的事情搞得很复杂,非要重写实现。
  
  我感觉这个应该很简单,有点不信邪,就自己来实现了。当然由于太久没编码,也踩了许多坑,东拼西凑也终于成功了,现在整理一下,方便后来人。
  鉴于我之前搜索资料时看到文章,目前感觉我这个是全网最简单的。。

重点:我做的是SSE通讯协议的MCP,因为studio的感觉用起来不如sse的广泛,url的实用性,验证也比较方便,使用了FastMCP框架。(有许多使用FastAPI的,但我感觉FastMCP这个最简单)
写了两个tools,一个是简单的回显工具,另一个是调用了api的工具,这样比较具有实战参考意义吧。

好了,进入正题:

一、配置环境

1、`sudo apt update  #工具更新`
2、apt install python3-pip #安装pip
3、pip install uv #此时有个报错
#报错信息:This environment is externally managed
#报错原因:表明Python环境是由操作系统的包管理器(如 apt、yum等)管理的,而不是由pip管理
#解决办法:创建虚拟环境,如下:
sudo apt install python3.12-venv #安装包
python3 -m venv mcp-env #创建虚拟
source mcp-env/bin/activate #激活虚拟环境
pip install uv #再次安装,成功

二、创建一个MCP的Demo

1、pip install fastmcp #安装fastmcp
2、uv run demo.py #启动,看到连接串
在这里插入图片描述
如果看到这个页面,就表示已经运行起来了,重点是看里面的transport和Server URL这两个内容,sse模式没错,URL改为正确的ip,就是一个AI真正可用的URL了。
3、来看业务代码:先写个最简单的回显

from fastmcp import FastMCP

# 仅2个必选参数:name + version
mcp  = FastMCP("sse-echo-server", "1.0")
# 回显工具
@mcp.tool
async def echo(params: dict):
    input_str = params.get("str", "默认字符串")
    echo_msg = f"回显结果:{input_str}"
    # 仅用框架最基础广播(不依赖任何transport子模块)
    try:
        await mcp.broadcast("echo_result", echo_msg)
    except:
        pass  # 若广播也不可用,至少工具调用能正常响应
    return {"code": 200, "data": echo_msg}
# 3. 启动
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

业务代码中的说明:其他的一看就能明白,唯独最后一个run的参数,花了比较久的时间(被元宝他们完全带偏,绕了不少弯路,几轮对话都越来越偏),最后还是得看看官方文档:https://gofastmcp.com/deployment/running-server
在这里插入图片描述
参照这个来写就好了,另外官方文档中,也写了如何使用studio和streamable的方式,有需要可以看着改下。

另外,对于启动命令,开始使用了fastmcp run xxx.py,结果发现这样的启动方式,只能启动studio模式,对于sse的,需要使用uv来启动。

三、测试

测试:
这个测试也很坑,因为我开始想用node的mcp调试工具来测试,结果耗费了许多不必要的时间,如果对这个不感兴趣,直接跳过,到与AI结合就可以了。

难点是这样的,我的MCP server运行在一台腾讯云的Light house上,使用的是ubuntu系统,所以它没有浏览器界面。因此,我要用本机的浏览器,访问我云服务器上web服务。

卡点就是:在云服务器上运行:

npx -y @modelcontextprotocol/inspector

在这里插入图片描述
是没有问题的,但是我本机通过浏览器访问 http://ip:6274时,就是访问不了,检查了防火墙,机器本身等一系列,也试了启动参数加上host 0.0.0.0,都不行,并没找到原因。

最最最后,只能去github上看readme了,还好,真发现了问题,在它的最后warning段落里,看到了说明:
在这里插入图片描述
重点是,这两个参数都要写才行,只写host,在控制台会报 非法来源的错误,且输入mcp信息后,连接不成功,报的错又很有误导性:说token问题。

总之:要这样写:

HOST=0.0.0.0 ALLOWED_ORIGINS=http://ip:6274,http://localhost:8000 npx -y @modelcontextprotocol/inspector

这样之后,就可以正常访问页面且能运行起来测试工具了。
在这里插入图片描述
测试成功。

四、集成

集成,可以使用的工具就比较多了,我拿了codebuddy和cusor两个工具来测试。都是填相同的内容即可
在这里插入图片描述
在这里插入图片描述

五、升级应用(再增加一个api调用)

使用过mcp的朋友都知道,在非常多的mcp server中,都有对于后端api的调用,以来保证信息的及时有效性,因此既然做demo,也就模拟个真实场景吧。

这部分的代码,就可以使用智能体来完成了,如使用codebuddy,输入:

用python3给我写一个应用:是一个在线api,输入参数是字符串,输出一个1-10的随机数字,运行端口是5000,带有web界面

它会为你生成一个代码,如下:

from flask import Flask, request, jsonify, render_template_string
import random

app = Flask(__name__)

@app.route('/random', methods=['GET'])
def get_random_number():
    input_string = request.args.get('input', '')
    random_number = random.randint(1, 10)
    return jsonify({
        'input': input_string,
        'random_number': random_number
    })

@app.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        input_string = request.form.get('input', '')
        random_number = random.randint(1, 10)
        return render_template_string('''
            <!DOCTYPE html>
            <html>
            <head>
                <title>Random Number Generator</title>
            </head>
            <body>
                <h1>Random Number Generator</h1>
                <form method="POST">
                    <label for="input">Enter a string:</label>
                    <input type="text" id="input" name="input" required>
                    <button type="submit">Generate</button>
                </form>
                {% if input %}
                    <p>Input: {{ input }}</p>
                    <p>Random Number: {{ random_number }}</p>
                {% endif %}
            </body>
            </html>
        ''', input=input_string, random_number=random_number)
    return render_template_string('''
        <!DOCTYPE html>
        <html>
        <head>
            <title>Random Number Generator</title>
        </head>
        <body>
            <h1>Random Number Generator</h1>
            <form method="POST">
                <label for="input">Enter a string:</label>
                <input type="text" id="input" name="input" required>
                <button type="submit">Generate</button>
            </form>
        </body>
        </html>
    ''')

if __name__ == '__main__':
    app.run(debug=True, port=5000)

可以运行这段,直接看到页面效果,验证功能正常。注:要在虚拟环境中,安装Flask,具体方式参照配置环境章节。

 配置虚拟环境并进入
 pip install flask
 python3 app.py

在这里插入图片描述
再通过AI,为我生成一个调用api的方法

再为我用python生成一个方法,名字是getRondom,来调用这个接口,使用localhost地址

又得到

import requests

def getRandom(input_string):
    """
    调用本地 API 获取随机数字
    :param input_string: 输入的字符串
    :return: API 响应结果(包含输入字符串和随机数字)
    """
    url = "http://127.0.0.1:5000/random"
    params = {"input": input_string}
    response = requests.get(url, params=params)
    return response.json()

# 示例用法
if __name__ == "__main__":
    result = getRandom("test")
    print(result)

找到其中有用的部分

import requests
def getRandom(input_string):
    """
    调用本地 API 获取随机数字
    :param input_string: 输入的字符串
    :return: API 响应结果(包含输入字符串和随机数字)
    """
    url = "http://127.0.0.1:5000/random"
    params = {"input": input_string}
    response = requests.get(url, params=params)
    return response.json()

把这段,也放在咱们的demo.py,最终文件是:

from fastmcp import FastMCP
import requests

# 1. 仅2个必选参数:name + version(无其他任何配置)
mcp  = FastMCP("sse-echo-server", "1.0")

# 2. 最简回显工具(函数名即工具名,无装饰器额外参数)
@mcp.tool
async def checkroute(params: dict):
    
    input_str = params.get("str", "没收到,显示默认字符串")
    echo_msg = f"回显结果:{input_str}"
    # 仅用框架最基础广播(不依赖任何transport子模块)
    try:
        await mcp.broadcast("echo_result", echo_msg)
    except:
        pass  # 若广播也不可用,至少工具调用能正常响应
    return {"code": 200, "data": echo_msg}

# 3. 随机数生成器(调用其他API)

@mcp.tool
async   def getRandom(input_string):
    """
    调用本地 API 获取随机数字
    :param input_string: 输入的字符串
    :return: API 响应结果(包含输入字符串和随机数字)
    """
    url = "http://127.0.0.1:5000/random"
    params = {"input": input_string}
    response = requests.get(url, params=params)
    return response.json()



# 3. 纯无参启动(完全依赖框架默认配置)
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=7777)

至此,已经完成了增加一个调用 api的tool,达到我们的效果。

六、最终测试

只要在codebuddy或cusor中刷新一下,就能看到到增加了tool,那就试试吧。
在这里插入图片描述

七、结语

有两点感受也分享下:
1、现在AI时代,许多应用,真的可以通过Agent来完成了,我们主要把握整体逻辑,懂一些基础知识,就能做过去可能许多天要做的事了,提效的效果确实立杆见影。

2、MCP当前也是热潮,各位也可以在MCP广场中,看到或者将自己企业的高质量MCP发布出来,让更多开发者集成使用,让更多用户在Agent里使用。

https://cloud.tencent.com/developer/mcp

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐