实战MCP协议:使用Gradio和Python构建可部署的AI工具链

引言:从理论到实践的跨越

当我们谈论AI应用开发时,往往会陷入两个极端:要么停留在概念层面,了解各种架构和协议却不知如何落地;要么直接使用复杂的框架,在配置文件和依赖关系的迷宫中迷失方向。

想象这样一个场景:你的团队正在开发一个客服系统,需要对用户反馈进行实时情感分析。传统的做法是什么?部署一个机器学习模型,编写API接口,然后在前端调用。但如果你想让这个能力被其他AI系统使用呢?你可能需要为每个AI平台编写不同的集成代码——为Claude Desktop写一套,为VS Code扩展写一套,为自定义Agent又写一套。

这正是MCP(模型上下文协议)要解决的问题。通过MCP,你只需要实现一次Server,任何支持MCP的Client都能无缝使用你的服务。听起来很美好,但具体怎么做?

本文将带你从零开始,构建一个完整的端到端MCP应用。我们将创建:

• • 一个基于Gradio的MCP Server(提供情感分析工具)
• • 多种Client实现(Python和JavaScript)
• • 一个可以使用这个工具的智能Agent
• • 部署到Hugging Face Spaces的完整流程

更重要的是,通过这个实战项目,你将真正理解MCP的工作机制,而不仅仅是停留在概念层面。让我们开始这段从理论到实践的旅程吧!

项目概览:我们要构建什么?

整体架构

我们将构建一个情感分析系统,整个架构分为三个主要部分:

┌─────────────────────────────────────────────┐
│           用户/AI Agent                      │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│         MCP Client (Tiny Agent)             │
│  - 连接到MCP Server                          │
│  - 发现可用工具                              │
│  - 调用情感分析工具                          │
└──────────────────┬──────────────────────────┘
                   │
                   ▼ (MCP协议 - JSON-RPC over HTTP/SSE)
┌─────────────────────────────────────────────┐
│      MCP Server (Gradio应用)                │
│  - 提供情感分析Tool                          │
│  - 处理文本分析请求                          │
│  - 返回情感得分和评估                        │
└─────────────────────────────────────────────┘

核心功能

Server端(Gradio):

• • 接收文本输入
• • 使用TextBlob库进行情感分析
• • 返回极性(polarity)、主观性(subjectivity)和情感评估
• • 同时提供Web UI和MCP API

Client端(Tiny Agent):

• • 自动发现Server提供的工具
• • 将工具提供给LLM使用
• • 处理LLM的工具调用请求
• • 支持多种推理提供商(Nebius、OpenAI等)

部署:

• • Server部署到Hugging Face Spaces
• • Client可以本地运行或部署为服务
• • 支持远程访问和协作

技术栈

Server开发:

• • Python 3.10+
• • Gradio (带MCP扩展)
• • TextBlob (情感分析库)

Client开发:

• • JavaScript: @huggingface/tiny-agents
• • Python: huggingface_hub[mcp]
• • Node.js (用于运行mcp-remote)

部署平台:

• • Hugging Face Spaces (免费托管)
• • Git (版本控制)

核心概念讲解

Gradio与MCP的完美结合

Gradio是一个流行的Python库,用于快速创建机器学习模型的Web界面。传统上,Gradio主要用于:

• • 快速原型开发
• • 模型演示
• • 数据可视化

但现在,Gradio加入了MCP支持,这意味着你可以用同一份代码同时实现:

1. 1. 人类友好的Web界面 - 用户可以直接在浏览器中使用
2. 2. AI友好的MCP接口 - AI模型可以通过标准协议调用

这种双重性质让Gradio成为构建MCP应用的理想选择。

Gradio的MCP自动转换机制

当你在Gradio中启用MCP Server时,它会自动完成以下转换:

1. 函数转换为Tool

def sentiment_analysis(text: str) -> str:
    """分析文本的情感倾向"""
    # 函数实现
    return result

自动转换为MCP Tool,名称为sentiment_analysis,描述来自docstring。

2. 参数转换为Schema

# Python类型提示
text: str

# 转换为JSON Schema
{
    "type": "string",
    "description": "文本输入"
}

3. 返回值格式化

Gradio会根据输出组件类型(Textbox、JSON等)自动格式化返回值。

Tiny Agents:极简的Agent实现

Tiny Agents是一个轻量级的Agent框架,核心理念是:

"Agent = LLM + Tools + While Loop"

这听起来简单,但这正是Agent的本质:

传统Agent框架的问题:

• • 过度工程化,学习曲线陡峭
• • 依赖众多,难以调试
• • 抽象层次太多,黑盒操作

Tiny Agents的优势:

• • 代码量极少(核心逻辑约50-70行)
• • 依赖简单,易于理解
• • 透明的执行流程
• • 支持多种LLM提供商

Tiny Agents的工作流程

1. 初始化Agent
   ↓
2. 连接到MCP Servers
   ↓
3. 发现可用Tools
   ↓
4. 进入对话循环:
   ├─ 接收用户输入
   ├─ 发送给LLM(附带Tool列表)
   ├─ LLM决定是否调用Tool
   ├─ 如果需要:执行Tool调用
   ├─ 将结果返回给LLM
   └─ LLM生成最终响应

MCP通信协议详解

在我们的应用中,Client和Server之间通过HTTP + SSE进行通信:

HTTP请求(Client → Server)

// 调用情感分析工具
POST /gradio_api/mcp/tools/call
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "sentiment_analysis",
    "arguments": {
      "text": "这个产品太棒了!"
    }
  }
}

SSE响应(Server → Client)

data: {
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{
      "type": "text",
      "text": "{\"polarity\": 0.85, \"subjectivity\": 0.6, \"assessment\": \"positive\"}"
    }]
  }
}

为什么使用SSE?

• • 支持Server到Client的实时推送
• • 比WebSocket更简单
• • 浏览器原生支持
• • 适合单向数据流场景

技术原理深入

Gradio MCP Server的内部机制

让我们深入了解Gradio如何将普通Python函数转换为MCP Tool。

步骤1:函数分析

当你调用demo.launch(mcp_server=True)时,Gradio会:

# 伪代码示例
def analyze_function(fn):
    # 提取函数签名
    signature = inspect.signature(fn)

    # 解析参数类型
    params = {}
    for param_name, param in signature.parameters.items():
        params[param_name] = {
            "type": python_type_to_json_type(param.annotation),
            "description": extract_from_docstring(fn.__doc__, param_name)
        }

    # 创建Tool schema
    tool_schema = {
        "name": fn.__name__,
        "description": fn.__doc__.split('\n')[0],
        "inputSchema": {
            "type": "object",
            "properties": params,
            "required": list(params.keys())
        }
    }

    return tool_schema

步骤2:端点注册

Gradio在以下URL注册MCP端点:

http://localhost:7860/gradio_api/mcp/
├── sse                    # SSE连接端点
├── schema                 # Tool schema查询
├── tools/list             # 工具列表
└── tools/call             # 工具调用

步骤3:请求处理

当收到工具调用请求时:

async def handle_tool_call(tool_name, arguments):
    # 1. 查找对应的函数
    fn = registered_functions[tool_name]

    # 2. 验证参数
    validate_arguments(fn, arguments)

    # 3. 执行函数
    result = await fn(**arguments)

    # 4. 格式化结果
    return format_mcp_response(result)

Tiny Agents的实现原理

让我们看看Tiny Agents如何实现Agent核心逻辑。

核心Agent循环(简化版)

class Agent:
    def __init__(self, model, servers):
        self.model = model
        self.tools = []
        # 连接到所有MCP servers并收集tools
        for server in servers:
            client = MCPClient(server)
            self.tools.extend(client.get_tools())

    async def run(self, user_message):
        messages = [{"role": "user", "content": user_message}]

        while True:
            # 1. 调用LLM
            response = await self.model.generate(
                messages=messages,
                tools=self.tools  # 提供可用工具
            )

            # 2. 检查是否需要调用工具
            if not response.tool_calls:
                # 没有工具调用,返回最终答案
                return response.content

            # 3. 执行工具调用
            for tool_call in response.tool_calls:
                result = await self.execute_tool(
                    tool_call.name,
                    tool_call.arguments
                )

                # 4. 将结果添加到对话历史
                messages.append({
                    "role": "tool",
                    "content": result,
                    "tool_call_id": tool_call.id
                })

            # 5. 继续循环,让LLM看到工具结果

MCP Client的实现

class MCPClient:
    def __init__(self, server_config):
        # 建立与Server的连接
        self.session = self.connect(server_config)

    async def get_tools(self):
        # 发送tools/list请求
        response = await self.session.request("tools/list", {})

        # 转换为Agent需要的格式
        tools = []
        for tool_schema in response["tools"]:
            tools.append({
                "type": "function",
                "function": {
                    "name": tool_schema["name"],
                    "description": tool_schema["description"],
                    "parameters": tool_schema["inputSchema"]
                }
            })

        return tools

    async def call_tool(self, name, arguments):
        # 发送tools/call请求
        response = await self.session.request("tools/call", {
            "name": name,
            "arguments": arguments
        })

        return response["content"][0]["text"]

连接流程详解

整个系统的初始化和运行流程如下:

1. Server启动
   ├─ Gradio启动Web服务器 (port 7860)
   ├─ 注册MCP端点
   ├─ 分析Python函数,生成Tool schema
   └─ 等待Client连接

2. Client初始化
   ├─ 读取agent配置文件(agent.json)
   ├─ 为每个Server创建MCP连接
   ├─ 发送initialize请求
   └─ Server返回协议版本和能力

3. Tool发现
   ├─ Client发送tools/list请求
   ├─ Server返回所有可用工具
   ├─ Client缓存工具列表
   └─ 准备就绪

4. 对话循环
   用户输入 → LLM推理 → 工具调用 → 结果返回 → LLM生成回复

代码实战

现在,让我们动手构建这个完整的应用。我们将分步骤完成Server和Client的开发。

第一步:搭建Gradio MCP Server

1.1 创建项目目录

# 创建项目文件夹
mkdir mcp-sentiment-app
cd mcp-sentiment-app

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate

1.2 安装依赖

# 安装Gradio(带MCP扩展)和TextBlob
pip install "gradio[mcp]" textblob

# 下载TextBlob的语料库
python -m textblob.download_corpora

依赖说明:

• • gradio[mcp]:Gradio核心库及MCP扩展
• • textblob:简单易用的自然语言处理库

1.3 编写Server代码

创建app.py文件(注意:Hugging Face Spaces要求使用这个文件名):

# app.py
import json
import gradio as gr
from textblob import TextBlob


def sentiment_analysis(text: str) -> str:
    """
    分析给定文本的情感倾向

    这个函数使用TextBlob库对输入文本进行情感分析,
    返回情感极性、主观性和总体评估。

    Args:
        text (str): 需要分析的文本内容

    Returns:
        str: 包含分析结果的JSON字符串,包含以下字段:
             - polarity: 情感极性,范围-1(消极)到1(积极)
             - subjectivity: 主观性,范围0(客观)到1(主观)
             - assessment: 情感评估(positive/negative/neutral)

    Example:
        >>> sentiment_analysis("这个产品太棒了!")
        '{"polarity": 0.8, "subjectivity": 0.7, "assessment": "positive"}'
    """
    # 创建TextBlob对象进行分析
    blob = TextBlob(text)
    sentiment = blob.sentiment

    # 构建结果字典
    result = {
        "polarity": round(sentiment.polarity, 2),
        "subjectivity": round(sentiment.subjectivity, 2),
        "assessment": (
            "positive" if sentiment.polarity > 0 else
            "negative" if sentiment.polarity < 0 else
            "neutral"
        )
    }

    # 返回JSON格式的字符串
    return json.dumps(result, ensure_ascii=False)


# 创建Gradio界面
demo = gr.Interface(
    fn=sentiment_analysis,                          # 要调用的函数
    inputs=gr.Textbox(                              # 输入组件
        placeholder="请输入要分析的文本...",
        lines=5,
        label="输入文本"
    ),
    outputs=gr.Textbox(                             # 输出组件
        label="分析结果",
        lines=3
    ),
    title="文本情感分析工具",
    description="使用TextBlob库分析文本的情感倾向,支持中英文",
    examples=[                                       # 示例输入
        ["这个电影太精彩了,强烈推荐!"],
        ["产品质量很差,非常失望。"],
        ["天气不错,适合出门。"]
    ],
    theme=gr.themes.Soft()                          # 使用柔和主题
)


# 启动应用
if __name__ == "__main__":
    # mcp_server=True 启用MCP Server功能
    demo.launch(
        mcp_server=True,
        server_name="0.0.0.0",  # 允许外部访问
        server_port=7860         # 默认端口
    )

1.4 代码详解

让我们逐部分理解这段代码:

情感分析函数:

def sentiment_analysis(text: str) -> str:

• • 类型提示:明确指定参数和返回值类型,Gradio会据此生成MCP schema
• • Docstring:提供详细文档,MCP会使用第一行作为工具描述
• • Args部分:描述每个参数,帮助LLM理解如何使用工具

TextBlob使用:

blob = TextBlob(text)
sentiment = blob.sentiment

TextBlob返回一个包含两个属性的对象:

• • polarity:情感极性,值域[-1, 1]
• • -1表示非常消极
• • 0表示中性
• • 1表示非常积极
• • subjectivity:主观性,值域[0, 1]
• • 0表示非常客观
• • 1表示非常主观

结果格式化:

return json.dumps(result, ensure_ascii=False)

• • 使用JSON格式便于程序化处理
• • ensure_ascii=False保留中文字符

Gradio配置:

demo = gr.Interface(
    fn=sentiment_analysis,
    inputs=gr.Textbox(...),
    outputs=gr.Textbox(...),  # 注意:使用Textbox而非JSON组件
    ...
)

重要:输出使用Textbox而非JSON,因为我们返回的是JSON字符串。

1.5 运行Server

python app.py

你会看到类似输出:

Running on local URL:  http://0.0.0.0:7860
Running MCP server at: http://0.0.0.0:7860/gradio_api/mcp/sse

To create a public link, set `share=True` in `launch()`.

测试Server:

1. 1. Web界面测试:
• • 打开 http://localhost:7860
• • 输入一些文本并提交
• • 查看分析结果
2. 2. MCP Schema查看:
• • 访问 http://localhost:7860/gradio_api/mcp/schema
• • 查看自动生成的Tool定义

你应该看到类似这样的schema:

{
  "tools": [{
    "name": "sentiment_analysis",
    "description": "分析给定文本的情感倾向",
    "inputSchema": {
      "type": "object",
      "properties": {
        "text": {
          "type": "string",
          "description": "需要分析的文本内容"
        }
      },
      "required": ["text"]
    }
  }]
}

第二步:构建Python Client(使用Tiny Agents)

现在Server已经运行,让我们创建一个Client来使用它。

2.1 安装依赖

# 安装huggingface_hub的MCP扩展
pip install "huggingface_hub[mcp]"

# 安装Node.js和npx(用于mcp-remote)
# 如果已安装可跳过
# 下载地址: https://nodejs.org/

# 安装mcp-remote包
npm install -g @modelcontextprotocol/mcp-remote

2.2 创建Agent配置文件

创建agent.json配置文件:

{
  "name": "sentiment-agent",
  "description": "情感分析助手,可以分析文本的情感倾向",
  "model": "Qwen/Qwen2.5-72B-Instruct",
  "provider": "nebius",
  "servers": [
    {
      "type": "stdio",
      "command": "npx",
      "args": [
        "@modelcontextprotocol/mcp-remote",
        "http://localhost:7860/gradio_api/mcp/sse"
      ]
    }
  ]
}

配置说明:

• • name:Agent的名称
• • description:Agent的功能描述
• • model:使用的LLM模型(这里用Qwen 2.5 72B)
• • provider:推理提供商(nebius提供免费API)
• • servers:要连接的MCP Server列表
• • type: "stdio":通过标准输入/输出通信
• • command:启动Server的命令
• • args:命令参数,这里使用mcp-remote作为代理

为什么需要mcp-remote?

一些MCP Client(如Claude Desktop)不直接支持HTTP+SSE的MCP Server,mcp-remote作为中间层,将HTTP+SSE转换为stdio协议。

2.3 设置API Token

Tiny Agents需要访问托管的LLM,需要设置Hugging Face token:

# 设置环境变量
export HF_TOKEN="your_huggingface_token_here"

# Windows CMD:
set HF_TOKEN=your_huggingface_token_here

# Windows PowerShell:
$env:HF_TOKEN="your_huggingface_token_here"

获取Token:

1. 1. 访问 https://huggingface.co/settings/tokens
2. 2. 创建一个新token(需要read权限)
3. 3. 复制token并设置到环境变量

2.4 运行Agent

# 使用CLI运行agent
tiny-agents run agent.json

交互示例:

🤖 Agent loaded with 1 tools:
  - sentiment_analysis: 分析给定文本的情感倾向

💬 You: 帮我分析一下这段话的情感:"今天天气真好,心情也很棒!"

🔧 Calling tool: sentiment_analysis
   Arguments: {"text": "今天天气真好,心情也很棒!"}

📊 Tool result: {"polarity": 0.75, "subjectivity": 0.6, "assessment": "positive"}

🤖 Assistant: 根据分析,这段话表达了积极正面的情感。具体来说:
- 情感极性为0.75,属于较强的积极情感
- 主观性为0.6,表明这是一个主观性较强的表达
- 总体评估为"positive"(积极)

这段话明显传达了愉悦和满意的情绪。

2.5 编程方式使用Agent

除了CLI,你也可以用Python代码直接使用Agent:

# client_example.py
import asyncio
import os
from huggingface_hub import Agent


async def main():
    # 创建Agent实例
    agent = Agent(
        model="Qwen/Qwen2.5-72B-Instruct",
        provider="nebius",
        api_key=os.getenv("HF_TOKEN"),
        servers=[
            {
                "type": "stdio",
                "command": "npx",
                "args": [
                    "@modelcontextprotocol/mcp-remote",
                    "http://localhost:7860/gradio_api/mcp/sse"
                ]
            }
        ]
    )

    # 加载工具
    await agent.load_tools()
    print(f"✅ 已加载 {len(agent.tools)} 个工具")

    # 运行对话
    user_input = "分析这句话的情感:这个产品质量太差了,非常失望。"
    print(f"\n💬 用户: {user_input}")

    response = await agent.run(user_input)
    print(f"\n🤖 助手: {response}")

    # 关闭连接
    await agent.cleanup()


# 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

运行这个脚本:

python client_example.py

第三步:构建JavaScript Client

如果你更喜欢JavaScript,这里是对应的实现。

3.1 初始化项目

# 创建新目录
mkdir js-client
cd js-client

# 初始化npm项目
npm init -y

# 安装依赖
npm install @huggingface/tiny-agents @modelcontextprotocol/mcp-remote

3.2 创建Agent代码

创建agent.js:

// agent.js
import { Agent } from '@huggingface/tiny-agents';


// 创建Agent实例
const agent = new Agent({
    // LLM配置
    model: "Qwen/Qwen2.5-72B-Instruct",
    provider: "nebius",
    apiKey: process.env.HF_TOKEN,

    // MCP Server配置
    servers: [
        {
            command: "npx",
            args: [
                "@modelcontextprotocol/mcp-remote",
                "http://localhost:7860/gradio_api/mcp/sse"
            ]
        }
    ]
});


// 主函数
async function main() {
    try {
        // 初始化Agent并加载工具
        await agent.initialize();
        console.log(`✅ Agent初始化完成`);
        console.log(`📦 可用工具: ${agent.tools.length}个\n`);

        // 列出所有工具
        agent.tools.forEach(tool => {
            console.log(`  - ${tool.name}: ${tool.description}`);
        });

        // 运行对话
        const userMessage = "帮我分析:这家餐厅的服务态度很好,但菜品一般。";
        console.log(`\n💬 用户: ${userMessage}`);

        const response = await agent.run(userMessage);
        console.log(`\n🤖 助手: ${response}`);

    } catch (error) {
        console.error("❌ 错误:", error.message);
    } finally {
        // 清理资源
        await agent.cleanup();
    }
}


// 运行
main();

3.3 更新package.json

添加ES模块支持:

{
  "name": "js-client",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "start": "node agent.js"
  },
  "dependencies": {
    "@huggingface/tiny-agents": "^latest",
    "@modelcontextprotocol/mcp-remote": "^latest"
  }
}

3.4 运行

export HF_TOKEN="your_token"
npm start

第四步:使用Gradio创建Client UI

我们还可以用Gradio创建一个可视化的Client界面:

4.1 创建Gradio Client

创建client_ui.py:

# client_ui.py
import gradio as gr
import os
from huggingface_hub import Agent
import asyncio


# 全局Agent实例
agent = None


async def initialize_agent():
    """初始化Agent并连接到MCP Server"""
    global agent

    if agent is None:
        agent = Agent(
            model="Qwen/Qwen2.5-72B-Instruct",
            provider="nebius",
            api_key=os.getenv("HF_TOKEN"),
            servers=[{
                "type": "stdio",
                "command": "npx",
                "args": [
                    "@modelcontextprotocol/mcp-remote",
                    "http://localhost:7860/gradio_api/mcp/sse"
                ]
            }]
        )
        await agent.load_tools()

    return agent


async def chat_with_agent(message, history):
    """
    与Agent对话

    Args:
        message: 用户输入
        history: 对话历史

    Returns:
        更新后的对话历史
    """
    # 确保Agent已初始化
    current_agent = await initialize_agent()

    # 显示用户消息
    history = history + [(message, None)]

    try:
        # 调用Agent
        response = await current_agent.run(message)

        # 更新历史记录
        history[-1] = (message, response)

    except Exception as e:
        error_message = f"❌ 错误: {str(e)}"
        history[-1] = (message, error_message)

    return history


def run_async_chat(message, history):
    """同步包装器,用于Gradio"""
    return asyncio.run(chat_with_agent(message, history))


# 创建Gradio聊天界面
with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🤖 情感分析AI助手")
    gr.Markdown("我可以帮你分析文本的情感倾向,支持中英文!")

    # 聊天界面
    chatbot = gr.Chatbot(
        label="对话",
        height=500,
        show_label=True
    )

    # 输入框
    msg = gr.Textbox(
        placeholder="输入你想分析的文本或问题...",
        label="消息",
        lines=2
    )

    # 按钮
    with gr.Row():
        submit = gr.Button("发送", variant="primary")
        clear = gr.Button("清空对话")

    # 示例
    gr.Examples(
        examples=[
            "分析这句话:这个电影太精彩了!",
            "这个产品质量很差,主观性和情感分别是多少?",
            "帮我看看'天气不错'是什么情感?"
        ],
        inputs=msg
    )

    # 事件绑定
    submit.click(
        run_async_chat,
        inputs=[msg, chatbot],
        outputs=[chatbot]
    )

    msg.submit(
        run_async_chat,
        inputs=[msg, chatbot],
        outputs=[chatbot]
    )

    clear.click(lambda: None, None, chatbot, queue=False)


# 启动
if __name__ == "__main__":
    demo.launch(share=False)

运行Client UI:

python client_ui.py

实际应用场景

场景1:社交媒体监控系统

**需求:**实时监控社交媒体上的品牌提及,分析公众情绪。

实现方案:

# social_monitor.py
import asyncio
from huggingface_hub import Agent


class SocialMediaMonitor:
    def __init__(self, mcp_server_url):
        self.agent = Agent(
            model="Qwen/Qwen2.5-72B-Instruct",
            provider="nebius",
            servers=[{
                "type": "stdio",
                "command": "npx",
                "args": ["@modelcontextprotocol/mcp-remote", mcp_server_url]
            }]
        )

    async def analyze_mentions(self, mentions):
        """批量分析社交媒体提及"""
        results = []

        for mention in mentions:
            prompt = f"分析这条评论的情感:{mention['text']}"
            response = await self.agent.run(prompt)

            results.append({
                "user": mention["user"],
                "text": mention["text"],
                "sentiment": response,
                "timestamp": mention["timestamp"]
            })

        return results

    async def generate_report(self, results):
        """生成情感分析报告"""
        positive = sum(1 for r in results if "positive" in r["sentiment"])
        negative = sum(1 for r in results if "negative" in r["sentiment"])
        neutral = sum(1 for r in results if "neutral" in r["sentiment"])

        report = f"""
        📊 情感分析报告
        ================
        总提及数: {len(results)}
        积极评论: {positive} ({positive/len(results)*100:.1f}%)
        消极评论: {negative} ({negative/len(results)*100:.1f}%)
        中性评论: {neutral} ({neutral/len(results)*100:.1f}%)
        """

        return report

场景2:客服工单自动分类

**需求:**根据客户反馈的情感,自动分配工单优先级。

实现方案:

# ticket_classifier.py
class TicketClassifier:
    def __init__(self, sentiment_agent):
        self.agent = sentiment_agent
        self.priority_rules = {
            "negative": {"polarity_threshold": -0.5, "priority": "high"},
            "neutral": {"priority": "medium"},
            "positive": {"priority": "low"}
        }

    async def classify_ticket(self, ticket_content):
        """分类客服工单"""
        # 使用Agent分析情感
        prompt = f"分析客户反馈的情感:{ticket_content}"
        sentiment_result = await self.agent.run(prompt)

        # 解析结果
        if "negative" in sentiment_result and "polarity" in sentiment_result:
            # 提取polarity值
            import re
            polarity_match = re.search(r'polarity.*?(-?\d+\.\d+)', sentiment_result)
            if polarity_match:
                polarity = float(polarity_match.group(1))

                # 确定优先级
                if polarity < -0.5:
                    priority = "urgent"  # 非常消极,紧急处理
                else:
                    priority = "high"
            else:
                priority = "high"
        elif "neutral" in sentiment_result:
            priority = "medium"
        else:
            priority = "low"

        return {
            "content": ticket_content,
            "sentiment": sentiment_result,
            "priority": priority,
            "auto_classified": True
        }

场景3:产品评论智能摘要

**需求:**从大量评论中提取关键情感信息,生成产品改进建议。

实现方案:

# review_summarizer.py
class ReviewSummarizer:
    def __init__(self, sentiment_agent):
        self.agent = sentiment_agent

    async def analyze_reviews(self, reviews):
        """分析多条评论"""
        sentiments = []

        for review in reviews:
            result = await self.agent.run(
                f"分析评论情感:{review}"
            )
            sentiments.append(result)

        return sentiments

    async def generate_insights(self, reviews, sentiments):
        """生成洞察报告"""
        # 构建综合分析提示
        combined_prompt = f"""
        基于以下评论和情感分析结果,生成产品改进建议:

        评论数量: {len(reviews)}
        情感分析结果: {sentiments}

        请总结:
        1. 主要的正面反馈点
        2. 主要的负面反馈点
        3. 改进建议
        """

        insights = await self.agent.run(combined_prompt)
        return insights

场景4:内容审核助手

**需求:**自动检测用户生成内容中的负面情绪,辅助人工审核。

实现方案:

# content_moderator.py
class ContentModerator:
    def __init__(self, sentiment_agent):
        self.agent = sentiment_agent
        self.risk_threshold = -0.7  # 情感极性阈值

    async def check_content(self, content):
        """检查内容是否需要人工审核"""
        # 分析情感
        prompt = f"详细分析这段内容的情感:{content}"
        result = await self.agent.run(prompt)

        # 判断是否需要审核
        needs_review = False
        risk_level = "low"

        if "negative" in result:
            # 提取polarity
            import json
            try:
                # 假设result中包含JSON格式数据
                if '"polarity"' in result:
                    import re
                    polarity_match = re.search(r'"polarity":\s*(-?\d+\.\d+)', result)
                    if polarity_match:
                        polarity = float(polarity_match.group(1))

                        if polarity < self.risk_threshold:
                            needs_review = True
                            risk_level = "high"
                        elif polarity < -0.3:
                            risk_level = "medium"
            except:
                pass

        return {
            "content": content,
            "sentiment_analysis": result,
            "needs_review": needs_review,
            "risk_level": risk_level
        }

部署到生产环境

部署到Hugging Face Spaces

步骤1:准备部署文件

在项目根目录创建以下文件:

requirements.txt

gradio[mcp]
textblob

README.md

---
title: Sentiment Analysis MCP Server
emoji: 🎭
colorFrom: blue
colorTo: green
sdk: gradio
sdk_version: 5.0.0
app_file: app.py
pinned: false
---

# 情感分析MCP Server

这是一个基于Gradio和MCP协议的情感分析服务。

## 功能

- 文本情感分析
- 支持中英文
- 提供Web UI和MCP API

## 使用方法

### Web界面
直接在页面中输入文本即可分析。

### MCP客户端
连接到以下端点:

https://YOUR_USERNAME-mcp-sentiment.hf.space/gradio_api/mcp/sse

步骤2:创建Space并推送代码
# 初始化git仓库
git init

# 添加文件
git add app.py requirements.txt README.md

# 提交
git commit -m "Initial commit: Sentiment Analysis MCP Server"

# 添加远程仓库
git remote add origin https://huggingface.co/spaces/YOUR_USERNAME/mcp-sentiment

# 推送代码
git push -u origin main

步骤3:配置Space

在Hugging Face Space设置中:

1. 1. 确认SDK为Gradio
2. 2. 确认app_file为app.py
3. 3. 等待构建完成

步骤4:测试部署

部署完成后,你的MCP Server将在以下地址可用:

https://YOUR_USERNAME-mcp-sentiment.hf.space/gradio_api/mcp/sse

更新你的agent配置:

{
  "servers": [{
    "type": "stdio",
    "command": "npx",
    "args": [
      "@modelcontextprotocol/mcp-remote",
      "https://YOUR_USERNAME-mcp-sentiment.hf.space/gradio_api/mcp/sse"
    ]
  }]
}

使用环境变量

对于生产环境,建议使用环境变量管理敏感信息:

在Gradio中:

import os

# 从环境变量读取配置
API_KEY = os.getenv("API_KEY")
MODEL_NAME = os.getenv("MODEL_NAME", "default-model")

demo.launch(
    mcp_server=True,
    auth=(os.getenv("USERNAME"), os.getenv("PASSWORD"))  # 添加认证
)

在Hugging Face Spaces中设置:

1. 1. 进入Space设置
2. 2. 找到"Repository secrets"
3. 3. 添加环境变量

监控和日志

添加日志记录功能:

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def sentiment_analysis(text: str) -> str:
    logger.info(f"收到分析请求,文本长度: {len(text)}")

    try:
        # 执行分析
        result = analyze(text)
        logger.info(f"分析完成,结果: {result['assessment']}")
        return json.dumps(result)

    except Exception as e:
        logger.error(f"分析失败: {str(e)}")
        raise

性能优化

1. 添加缓存:

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_sentiment_analysis(text: str) -> str:
    """带缓存的情感分析"""
    return sentiment_analysis(text)

2. 批处理支持:

def batch_sentiment_analysis(texts: list) -> list:
    """批量分析情感"""
    results = []
    for text in texts:
        results.append(sentiment_analysis(text))
    return results

3. 异步处理:

import asyncio

async def async_sentiment_analysis(text: str) -> str:
    """异步版本的情感分析"""
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        sentiment_analysis,
        text
    )
    return result

常见问题和解决方案

问题1:连接失败

症状:

Error: Failed to connect to MCP server

可能原因和解决方案:

1. 1. Server未运行

   # 检查Server是否在运行
   curl http://localhost:7860/gradio_api/mcp/schema
   
   # 如果没有响应,启动Server
   python app.py

2. 2. 端口被占用

   # 更改端口
   demo.launch(mcp_server=True, server_port=7861)

3. 3. 防火墙阻止

   # 允许端口访问(Linux)
   sudo ufw allow 7860

问题2:工具未被发现

症状:
Agent显示"No tools available"

解决方案:

1. 1. 检查函数定义

   # 确保有完整的类型提示和docstring
   def my_tool(arg: str) -> str:
       """工具描述"""
       return result

2. 2. 检查MCP schema

   # 访问schema端点
   curl http://localhost:7860/gradio_api/mcp/schema

3. 3. 重启Server和Client

   # 完全重启
   pkill -f "python app.py"
   python app.py

问题3:工具调用超时

症状:

TimeoutError: Tool call exceeded timeout limit

解决方案:

1. 1. 增加超时时间

   # 在Agent配置中
   agent = Agent(
       ...
       timeout=60  # 增加到60秒
   )

2. 2. 优化函数性能

   # 添加缓存
   @lru_cache(maxsize=100)
   def sentiment_analysis(text: str):
       ...

3. 3. 使用异步

   async def async_tool(text: str):
       # 异步处理避免阻塞
       ...

问题4:中文显示乱码

解决方案:

import json

# 使用ensure_ascii=False
result = json.dumps(data, ensure_ascii=False)

# 或者设置环境变量
import os
os.environ['PYTHONIOENCODING'] = 'utf-8'

问题5:Gradio更新导致不兼容

解决方案:

# 锁定Gradio版本
pip install "gradio[mcp]==5.0.0"

# 或在requirements.txt中指定
gradio[mcp]==5.0.0

总结

通过本文的完整实战,我们从零构建了一个端到端的MCP应用。让我们回顾关键要点:

核心收获:

1. 1. Gradio简化开发:通过mcp_server=True,一行代码即可将Python函数转换为MCP Tool,大大降低了Server开发难度
2. 2. Tiny Agents轻量高效:用极简的代码实现了完整的Agent功能,证明了"简单即是美"的设计哲学
3. 3. 端到端可行性:从Server开发、Client连接到生产部署,整个流程是完全可行且易于实现的
4. 4. 生态系统优势:基于Hugging Face生态,我们可以轻松访问各种模型、数据集和部署服务