开篇：当「大脑」遇见「工具手」🧠✋

想象一下：

• AI Agent = 聪明的大脑，会思考、会规划

• FastMCP = 灵巧的双手，能操作各种工具

两者结合 = 一个既聪明又能干的超级AI！🤖💪

传统的AI Agent只能「空想」，而有了FastMCP，它就能真正「动手」操作你的电脑、访问你的数据、执行真实的任务！

---

一、🎯 为什么需要结合？优势分析

场景	只有AI Agent	AI Agent + FastMCP
文件操作	❌ 只能描述如何操作	✅ 实际创建、编辑、搜索文件
数据分析	❌ 只能给出分析思路	✅ 直接读取数据库进行分析
系统管理	❌ 只能提供命令示例	✅ 实际执行系统命令
实时信息	❌ 依赖训练数据（可能过时）	✅ 获取实时数据和信息

---

二、🛠️ 架构设计：如何结合？

三、🚀 实战Demo：构建一个自动化任务Agent

我们将创建一个 「智能项目助手」，它能够：

1. 分析项目需求

2. 自动创建项目文件结构

3. 生成基础代码

4. 管理项目任务

第1步：安装必要的库

pip install fastmcp langchain-openai langchain-agents python-dotenv requests

第2步：创建FastMCP Server（工具层）

创建 project_tools_server.py：

# project_tools_server.py
from fastmcp import FastMCP, Tool
import os
import json
from pathlib import Path
from typing import List, Dict, Any
import subprocess
import glob

# 初始化FastMCP Server
mcp = FastMCP("project-assistant", description="智能项目助手工具集")

@mcp.tool()
def create_project_structure(project_name: str, structure: Dict[str, Any]) -> str:
    """创建项目文件结构
    
    Args:
        project_name: 项目名称
        structure: 项目结构定义，支持嵌套
    """
    base_path = Path(project_name)
    base_path.mkdir(exist_ok=True)
    
    def create_structure(current_path: Path, current_structure: Dict[str, Any]):
        for name, content in current_structure.items():
            item_path = current_path / name
            
            if isinstance(content, dict):
                # 如果是目录
                item_path.mkdir(exist_ok=True)
                create_structure(item_path, content)
            else:
                # 如果是文件
                item_path.write_text(content, encoding='utf-8')
    
    create_structure(base_path, structure)
    return f"✅ 项目 '{project_name}' 结构创建完成！"

@mcp.tool()
def analyze_directory(path: str = ".") -> str:
    """分析当前目录的文件结构
    
    Args:
        path: 要分析的路径，默认为当前目录
    """
    analysis = {"files": 0, "directories": 0, "file_types": {}}
    
    def analyze_path(current_path: Path):
        for item in current_path.iterdir():
            if item.name.startswith('.'):
                continue
                
            if item.is_dir():
                analysis["directories"] += 1
                analyze_path(item)
            else:
                analysis["files"] += 1
                ext = item.suffix.lower()
                analysis["file_types"][ext] = analysis["file_types"].get(ext, 0) + 1
    
    analyze_path(Path(path))
    
    result = f"📊 目录分析结果: {path}\n\n"
    result += f"📁 目录数: {analysis['directories']}\n"
    result += f"📄 文件数: {analysis['files']}\n\n"
    result += "📋 文件类型统计:\n"
    for ext, count in analysis["file_types"].items():
        result += f"  {ext or '无扩展名'}: {count} 个\n"
    
    return result

@mcp.tool()
def create_file(filepath: str, content: str = "") -> str:
    """创建或编辑文件
    
    Args:
        filepath: 文件路径
        content: 文件内容
    """
    path = Path(filepath)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding='utf-8')
    return f"✅ 文件 '{filepath}' 创建成功！"

@mcp.tool()
def read_file(filepath: str) -> str:
    """读取文件内容
    
    Args:
        filepath: 文件路径
    """
    try:
        content = Path(filepath).read_text(encoding='utf-8')
        return f"📄 文件内容: {filepath}\n\n{content}"
    except FileNotFoundError:
        return f"❌ 文件不存在: {filepath}"
    except Exception as e:
        return f"❌ 读取文件失败: {str(e)}"

@mcp.tool()
def run_command(command: str, cwd: str = ".") -> str:
    """执行系统命令
    
    Args:
        command: 要执行的命令
        cwd: 工作目录
    """
    try:
        result = subprocess.run(
            command, 
            shell=True, 
            capture_output=True, 
            text=True, 
            cwd=cwd
        )
        output = f"🚀 命令: {command}\n\n"
        if result.stdout:
            output += f"📤 输出:\n{result.stdout}\n"
        if result.stderr:
            output += f"📣 错误:\n{result.stderr}\n"
        output += f"🔚 返回码: {result.returncode}"
        return output
    except Exception as e:
        return f"❌ 执行命令失败: {str(e)}"

@mcp.tool()
def find_files(pattern: str, path: str = ".") -> str:
    """查找匹配模式的文件
    
    Args:
        pattern: 文件模式（如 *.py, *.md）
        path: 搜索路径
    """
    files = glob.glob(f"{path}/**/{pattern}", recursive=True)
    if not files:
        return f"🔍 没有找到匹配 '{pattern}' 的文件"
    
    result = f"📁 找到 {len(files)} 个匹配 '{pattern}' 的文件:\n\n"
    for file in files:
        result += f"  • {file}\n"
    return result

if __name__ == "__main__":
    mcp.run(transport="stdio")

第3步：创建AI Agent（大脑层）

创建 smart_agent.py：

# smart_agent.py
import os
import json
from langchain.agents import AgentType, initialize_agent
from langchain_openai import OpenAI
from langchain.schema import SystemMessage
from langchain.agents import Tool
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client
import asyncio
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class FastMCPClient:
    """FastMCP客户端封装类"""
    
    def __init__(self):
        self.tools = []
        self.available_functions = {}
    
    async def connect(self):
        """连接到FastMCP Server"""
        try:
            self.reader, self.writer = await stdio_client("python", "project_tools_server.py")
            self.session = ClientSession(self.reader, self.writer)
            await self.session.initialize()
            
            # 获取可用的工具
            tools_response = await self.session.list_tools()
            self.tools = tools_response.tools
            
            # 创建工具映射
            self.available_functions = {
                tool.name: self.create_tool_func(tool.name) 
                for tool in self.tools
            }
            
            print("✅ FastMCP连接成功！")
            print(f"🛠️  可用工具: {[tool.name for tool in self.tools]}")
            
        except Exception as e:
            print(f"❌ 连接FastMCP失败: {e}")
            raise
    
    def create_tool_func(self, tool_name):
        """创建工具函数"""
        async def tool_function(**kwargs):
            try:
                result = await self.session.call_tool(tool_name, kwargs)
                return result.content[0].text
            except Exception as e:
                return f"❌ 工具调用失败: {str(e)}"
        return tool_function
    
    async def close(self):
        """关闭连接"""
        if hasattr(self, 'session'):
            await self.session.close()
        if hasattr(self, 'writer'):
            self.writer.close()

# 初始化AI Agent
async def create_agent():
    # 创建FastMCP客户端并连接
    mcp_client = FastMCPClient()
    await mcp_client.connect()
    
    # 初始化LLM
    llm = OpenAI(
        model_name="gpt-3.5-turbo-instruct",
        temperature=0.3
    )
    
    # 创建LangChain工具
    langchain_tools = []
    for tool in mcp_client.tools:
        langchain_tools.append(
            Tool(
                name=tool.name,
                func=lambda input_str, tool_name=tool.name: asyncio.run(
                    mcp_client.available_functions[tool_name](**json.loads(input_str))
                ),
                description=tool.description,
            )
        )
    
    # 创建系统提示词
    system_message = SystemMessage(content="""你是一个智能项目助手，拥有强大的工具能力。
你可以帮助用户创建项目、分析代码、执行命令、管理文件等。
请根据用户需求选择合适的工具，并给出详细的执行结果。""")
    
    # 初始化Agent
    agent = initialize_agent(
        tools=langchain_tools,
        llm=llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True,
        handle_parsing_errors=True,
        agent_kwargs={
            "system_message": system_message
        }
    )
    
    return agent, mcp_client

# 主程序
async def main():
    print("🚀 正在启动智能项目助手...")
    
    try:
        # 创建Agent和MCP客户端
        agent, mcp_client = await create_agent()
        
        print("\n🤖 智能助手已就绪！")
        print("你可以让我：")
        print("  - 创建新项目并设置文件结构")
        print("  - 分析当前目录的文件")
        print("  - 执行系统命令")
        print("  - 查找特定类型的文件")
        print("  - 创建和编辑文件")
        print("输入 'quit' 退出程序")
        
        while True:
            user_input = input("\n👨💻 你: ")
            if user_input.lower() in ['quit', 'exit', 'q']:
                break
            
            try:
                # 运行Agent
                result = await agent.arun(user_input)
                print(f"\n🤖 助手: {result}")
                
            except Exception as e:
                print(f"❌ 执行出错: {str(e)}")
    
    finally:
        # 关闭连接
        if 'mcp_client' in locals():
            await mcp_client.close()
        print("👋 程序已退出")

if __name__ == "__main__":
    asyncio.run(main())

第4步：创建配置文件

创建 .env 文件：

OPENAI_API_KEY=你的OpenAI_API_KEY

第5步：运行智能助手！

python smart_agent.py

四、🎯 实际使用示例

示例1：创建Python项目

👨💻 你: 帮我创建一个名为my_app的Python项目，包含main.py、requirements.txt和README.md文件

🤖 助手: 正在创建项目结构...
✅ 项目 'my_app' 结构创建完成！

📊 新建的项目包含：
• my_app/main.py - 主程序文件
• my_app/requirements.txt - 依赖文件  
• my_app/README.md - 项目说明文档

示例2：分析项目并安装依赖

👨💻 你: 分析一下my_app目录，然后安装Flask框架

🤖 助手: 
📊 目录分析结果: my_app
📁 目录数: 1
📄 文件数: 3

🚀 正在执行: pip install flask
✅ Flask安装成功！

示例3：创建复杂的项目结构

👨💻 你: 帮我创建一个Web项目，包含static/css、templates目录和app.py主文件

🤖 助手: 
✅ 创建了完整的Web项目结构！
📁 my_web_project/
  ├── static/
  │   └── css/
  ├── templates/
  └── app.py

五、💡 高级用法：自定义工具扩展

你还可以轻松添加更多专业工具：

# 在FastMCP Server中添加更多工具
@mcp.tool()
def code_analysis(filepath: str) -> str:
    """代码质量分析"""
    # 实现代码检查逻辑
    return "代码分析报告..."

@mcp.tool() 
def git_operation(operation: str, message: str = "") -> str:
    """Git操作"""
    # 实现Git命令执行
    return "Git操作完成"

@mcp.tool()
def database_query(query: str, db_path: str) -> str:
    """数据库查询"""
    # 实现SQLite或其他数据库查询
    return "查询结果..."

六、🎯 架构优势总结

1. 分离关注点: Agent负责思考，MCP负责执行

2. 安全性: 工具操作在受控环境中进行

3. 可扩展性: 可以轻松添加新的工具

4. 复用性: 同样的工具可以被不同的Agent使用

5. 标准化: 遵循MCP协议，兼容各种客户端

---

七、🚀 生产环境部署建议

对于生产环境，你可以：

使用Docker容器化：

FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "smart_agent.py"]

添加身份验证：

# 在FastMCP Server中添加认证
mcp = FastMCP("project-assistant", require_auth=True)

实现日志监控：

import logging
logging.basicConfig(level=logging.INFO)

总结：🤖 + 🛠️ = 无限可能

通过将 AI Agent 与 FastMCP 结合，你创建的不再是一个只能聊天的AI，而是一个真正能够理解需求、规划任务、执行操作的智能助手！

这种组合开启了无限可能：

• 🏗️ 自动化项目搭建

• 📊 智能数据分析

• 🔧 系统管理自动化

• 🚀 DevOps流程优化

现在就去尝试构建你自己的「AI员工」吧！让你的AI助手从「顾问」升级为「执行者」！💪