在这篇文章中，我们将探讨如何使用FastAPI构建一个MCP服务器，并将其与谷歌的Gemini AI集成，以便从文件中处理文本内容。这个项目实现了按需从文件中提取文本内容，并利用生成式AI来进行格式化和摘要。

这个项目主要有两部分：

1. MCP Server, 一个使用FastAPI构建的服务，可以从各种文件格式中读取文本。
2. 客户端, 从MCP服务器获取文本，并用Gemini AI处理。

技术堆栈

• FastAPI（用于MCP服务器的框架）
• MCP协议（用于获取结构化数据）
• Google Gemini AI（用于生成文本摘要和格式化文本）
• httpx（用于发送异步HTTP请求）
• Uvicorn（用于运行FastAPI服务器的工具）
• Python Asyncio（用于处理异步任务）

MCP服务器提供了一个API端点来读取文件中的文本。它支持多种文件格式，如.txt、.json、.xml、.csv、.docx等。

    pip install fastapi uvicorn httpx aiofiles pandas python-docx

    from typing import Union  
    import os  
    import csv  
    import json  
    import xml.etree.ElementTree as ET  
    from fastapi import FastAPI  
    from docx import Document  
    import uvicorn  

    app = FastAPI()  
    @app.get("/read-text-from-file")  
    def read_text_from_file(file_path: str) -> Union[str, None]:  
        try:  
            if not os.path.exists(file_path):  
                print(f"找不到文件: {file_path}")  
                return None  
            _, file_ext = os.path.splitext(file_path)  
            if file_ext in (".txt", ".log", ".md"):  
                with open(file_path, "r", encoding="utf-8") as f:  
                    return f.read()  
            elif file_ext == ".json":  
                with open(file_path, "r", encoding="utf-8") as f:  
                    return json.dumps(json.load(f), indent=4, ensure_ascii=False)  
            elif file_ext == ".xml":  
                tree = ET.parse(file_path)  
                return ET.tostring(tree.getroot(), encoding="unicode")  
            elif file_ext in (".csv", ".tsv"):  
                delimiter = "," if file_ext == ".csv" else "\t"  
                with open(file_path, "r", encoding="utf-8") as f:  
                    reader = csv.reader(f, delimiter=delimiter)  
                    return "\n".join([delimiter.join(row) for row in reader])  
            elif file_ext in (".docx", ".doc"):  
                return "\n".join([p.text for p in Document(file_path).paragraphs])  
            else:  
                print(f"不支持的文件类型: {file_ext}")  
                return None  
        except Exception as e:  
            print(f"读取文件时出错: {e}")  
            return None  
    if __name__ == "__main__":  
        uvicorn.run(app, host="127.0.0.1", port=8000)

客户端从MCP服务器取回文本，并将其发送到Google的Gemini API进行格式化和总结。

pip install google-generativeai python-dotenv httpx

import json  
import httpx  
import google.generativeai as genai  
import os  
from dotenv import load_dotenv  

load_dotenv()  
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://127.0.0.1:8000")  
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")  
genai.configure(api_key=GEMINI_API_KEY)  
async def fetch_text_from_mcp(file_path: str):  
    async with httpx.AsyncClient() as client:  
        response = await client.get(  
            f"{MCP_SERVER_URL}/read-text-from-file",  
            params={"file_path": file_path},  
        )  
        if response.status_code == 200:  
            return response.json() or "未能成功获取数据"  
        else:  
            return f"错误代码: {response.status_code}, {response.text}"  
def generate_gemini_response(text: str):  
    model = genai.GenerativeModel("gemini-1.5-pro")  
    response = model.generate_content(f"请格式化并总结以下内容:\n{text}")  
    return response.text  
async def main():  
    file_path = "data.txt"  
    text_content = await fetch_text_from_mcp(file_path)  
    print('文本内容--------:', text_content)  
    if text_content:  
        formatted_response = generate_gemini_response(text_content)  
        print("=== 处理结果 ===")  
        print(formatted_response)  
    else:  
        print("未能读取文件。")  
if __name__ == "__main__":  
    import asyncio  
    asyncio.run(main())

1. 启动MCP服务器：

python mcp_server.py  # 运行MCP服务器的Python脚本

1. 运行客户端应用。

在命令行中输入以下命令来运行主程序：

python main.py

• Docker化: 作为容器化服务部署。
• 云端部署: 在Google Cloud Run或AWS Lambda上运行MCP服务器。
• 多文件处理: 支持多文件处理请求。
• 日志和监控: 实现更优秀的日志记录，使用结构化日志。

这个项目展示了如何将MCP服务器与生成式AI模型集成，以抓取、处理并总结多种文件格式中的文本。通过进一步改进，这可以扩展为一个强大的基于AI的文档处理系统。
完整源代码请访问GitHub仓库页面：Model-Context-Protocol 项目页面。

有任何问题或建议，随时告诉我哦！🚀