项目结构

mcp-langgraph-demo/
├── client/
│   ├── main.py
│   ├── mcp_client.py
│   └── graph_builder.py
├── server/
│   ├── app.py
│   ├── mcp_server.py
│   └── knowledge_base.py
├── requirements.txt
└── README.md

步骤一：服务端实现 (SSE + IO模式)

1. 服务端核心代码

# server/mcp_server.py
from flask import Flask, request, jsonify, Response
import json
import time
from typing import List, Dict, Any
import threading

app = Flask(__name__)

# 模拟知识库
KNOWLEDGE_BASE = [
    {
        "id": "1",
        "text": "Python 装饰器是一种特殊类型的函数，它可以修改其他函数的功能。装饰器本质上是一个接受函数作为参数并返回函数的函数。",
        "source": "Python 基础教程",
        "tags": ["python", "decorator"],
        "score": 0.95
    },
    {
        "id": "2", 
        "text": "常见的装饰器包括 @property、@staticmethod、@classmethod 等。自定义装饰器可以用来实现日志记录、性能测试、事务处理等功能。",
        "source": "Python 高级编程",
        "tags": ["python", "decorator", "advanced"],
        "score": 0.88
    },
    {
        "id": "3",
        "text": "数据库查询优化的关键在于合理使用索引、避免全表扫描、优化 SQL 语句结构。常见的优化策略包括：添加合适的索引、使用 EXPLAIN 分析查询计划、避免 SELECT * 等。",
        "source": "数据库性能优化指南",
        "tags": ["database", "optimization", "sql"],
        "score": 0.92
    },
    {
        "id": "4",
        "text": "React Hooks 是 React 16.8 引入的新特性，允许在函数组件中使用状态和其他 React 特性。相比 Class 组件，Hooks 提供了更简洁的代码结构和更好的逻辑复用能力。",
        "source": "React 官方文档",
        "tags": ["react", "hooks", "frontend"],
        "score": 0.90
    },
    {
        "id": "5",
        "text": "Docker 容器化部署的主要优势包括：环境一致性、快速部署、资源隔离、易于扩展、简化运维等。容器化使得应用可以在不同环境中保持一致的运行表现。",
        "source": "Docker 实践指南",
        "tags": ["docker", "devops", "deployment"],
        "score": 0.87
    }
]

class MCPServer:
    @staticmethod
    def search(query: str, limit: int = 5) -> List[Dict[str, Any]]:
        """搜索相关文档"""
        results = []
        query_lower = query.lower()
        
        for doc in KNOWLEDGE_BASE:
            # 简单的相关性计算
            score = 0
            if query_lower in doc["text"].lower():
                score = doc["score"]
            elif any(tag in query_lower for tag in doc["tags"]):
                score = doc["score"] * 0.8
            
            if score > 0:
                results.append({
                    "id": doc["id"],
                    "text": doc["text"],
                    "meta": {
                        "source": doc["source"],
                        "tags": doc["tags"]
                    },
                    "score": round(score, 2)
                })
        
        # 按分数排序
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:limit]

# IO 模式接口
@app.route('/api/search', methods=['POST'])
def search_sync():
    """同步搜索接口 (IO模式)"""
    try:
        data = request.get_json()
        query = data.get('query', '')
        limit = data.get('limit', 5)
        
        if not query:
            return jsonify({"error": "Query is required"}), 400
        
        results = MCPServer.search(query, limit)
        return jsonify({
            "query": query,
            "results": results,
            "total": len(results)
        })
    
    except Exception as e:
        return jsonify({"error": str(e)}), 500

# SSE 模式接口
@app.route('/api/search/stream', methods=['POST'])
def search_stream():
    """流式搜索接口 (SSE模式)"""
    def generate():
        try:
            data = request.get_json()
            query = data.get('query', '')
            limit = data.get('limit', 5)
            
            if not query:
                yield f"data: {json.dumps({'error': 'Query is required'})}\n\n"
                return
            
            # 发送开始信号
            yield f"data: {json.dumps({'type': 'start', 'query': query})}\n\n"
            
            # 模拟逐步返回结果
            results = MCPServer.search(query, limit)
            
            for i, result in enumerate(results):
                # 模拟延迟，展示流式效果
                time.sleep(0.5)
                yield f"data: {json.dumps({'type': 'result', 'index': i, 'data': result})}\n\n"
            
            # 发送结束信号
            yield f"data: {json.dumps({'type': 'end', 'total': len(results)})}\n\n"
            
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
    
    return Response(generate(), mimetype='text/plain')

@app.route('/health', methods=['GET'])
def health_check():
    """健康检查"""
    return jsonify({"status": "healthy", "service": "MCP Server"})

if __name__ == '__main__':
    print("🚀 MCP Server 启动中...")
    print("服务地址: http://localhost:5000")
    print("同步接口: POST /api/search")
    print("流式接口: POST /api/search/stream")
    app.run(host='localhost', port=5000, debug=True)

步骤二：客户端实现

1. MCP 客户端

# client/mcp_client.py
import aiohttp
import asyncio
import json
from typing import List, Dict, Any, AsyncGenerator
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MCPClient:
    def __init__(self, base_url: str = "http://localhost:5000"):
        self.base_url = base_url.rstrip('/')
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
    
    async def search_io(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
        """IO模式搜索 - 同步等待完整结果"""
        if not self.session:
            raise RuntimeError("MCPClient must be used within async context manager")
        
        try:
            async with self.session.post(
                f"{self.base_url}/api/search",
                json={
                    "query": query,
                    "limit": limit
                },
                timeout=aiohttp.ClientTimeout(total=30)
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get("results", [])
                else:
                    error_text = await response.text()
                    logger.error(f"MCP search failed: {response.status} - {error_text}")
                    return []
        except Exception as e:
            logger.error(f"MCP search error: {e}")
            return []
    
    async def search_sse(self, query: str, limit: int = 5) -> AsyncGenerator[Dict, None]:
        """SSE模式搜索 - 流式接收结果"""
        if not self.session:
            raise RuntimeError("MCPClient must be used within async context manager")
        
        try:
            async with self.session.post(
                f"{self.base_url}/api/search/stream",
                json={
                    "query": query,
                    "limit": limit
                },
                timeout=aiohttp.ClientTimeout(total=30)
            ) as response:
                if response.status == 200:
                    async for line in response.content:
                        line_str = line.decode('utf-8').strip()
                        if line_str.startswith('data: '):
                            data_str = line_str[6:]  # 移除 'data: ' 前缀
                            try:
                                event_data = json.loads(data_str)
                                yield event_data
                            except json.JSONDecodeError:
                                logger.warning(f"Failed to parse JSON: {data_str}")
                else:
                    error_text = await response.text()
                    logger.error(f"MCP stream search failed: {response.status} - {error_text}")
                    yield {"type": "error", "message": f"HTTP {response.status}: {error_text}"}
        except Exception as e:
            logger.error(f"MCP stream search error: {e}")
            yield {"type": "error", "message": str(e)}
    
    def format_context(self, items: List[Dict]) -> str:
        """格式化上下文为可读文本"""
        if not items:
            return "未找到相关上下文信息。"
        
        formatted_items = []
        for i, item in enumerate(items, 1):
            text = item.get("text", "")
            source = item.get("meta", {}).get("source", "未知来源")
            score = item.get("score", 0)
            
            formatted_item = f"{i}. {text}\n   来源: {source} (相关度: {score:.2f})"
            formatted_items.append(formatted_item)
        
        return "\n\n".join(formatted_items)

2. LangGraph 状态图

# client/graph_builder.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, AsyncGenerator
import operator
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import asyncio

class GraphState(TypedDict):
    messages: Annotated[list, operator.add]
    context: str
    answer: str
    next_step: str
    mode: str  # 'io' 或 'sse'
    streaming_results: list

class BlogQAGraph:
    def __init__(self, mcp_client, llm=None):
        self.mcp_client = mcp_client
        self.llm = llm or ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        self.graph = self._build_graph()
    
    def _build_graph(self):
        """构建状态图"""
        workflow = StateGraph(GraphState)
        
        # 添加节点
        workflow.add_node("retrieve_context", self._retrieve_context)
        workflow.add_node("generate_answer", self._generate_answer)
        workflow.add_node("validate_answer", self._validate_answer)
        
        # 设置流程
        workflow.set_entry_point("retrieve_context")
        workflow.add_edge("retrieve_context", "generate_answer")
        workflow.add_edge("generate_answer", "validate_answer")
        workflow.add_edge("validate_answer", END)
        
        return workflow.compile()
    
    async def _retrieve_context(self, state: GraphState) -> GraphState:
        """检索上下文阶段"""
        try:
            query = state["messages"][-1].content
            mode = state.get("mode", "io")
            
            print(f"🔍 [{mode.upper()}模式] 正在检索关于 '{query}' 的上下文...")
            
            if mode == "io":
                # IO模式：等待完整结果
                context_items = await self.mcp_client.search_io(query)
                context_text = self.mcp_client.format_context(context_items)
                print(f"📚 [{mode.upper()}模式] 检索到 {len(context_items)} 条相关上下文")
                
                return {
                    "context": context_text,
                    "next_step": "generate_answer",
                    "streaming_results": []
                }
            
            elif mode == "sse":
                # SSE模式：流式处理
                streaming_results = []
                async for event in self.mcp_client.search_sse(query):
                    if event["type"] == "result":
                        streaming_results.append(event["data"])
                        print(f"📥 [SSE模式] 接收到结果 {event['index'] + 1}")
                    elif event["type"] == "end":
                        print(f"🏁 [SSE模式] 检索完成，共 {event['total']} 条结果")
                    elif event["type"] == "error":
                        print(f"❌ [SSE模式] 检索错误: {event['message']}")
                
                context_text = self.mcp_client.format_context(streaming_results)
                
                return {
                    "context": context_text,
                    "next_step": "generate_answer",
                    "streaming_results": streaming_results
                }
            
        except Exception as e:
            error_msg = f"检索上下文时出错: {str(e)}"
            print(f"❌ {error_msg}")
            return {
                "context": "",
                "next_step": "generate_answer",
                "streaming_results": [],
                "error": error_msg
            }
    
    def _generate_answer(self, state: GraphState) -> GraphState:
        """生成回答阶段"""
        try:
            query = state["messages"][-1].content
            context = state["context"]
            mode = state.get("mode", "io")
            
            print(f"🤖 [{mode.upper()}模式] 正在生成回答...")
            
            # 构建提示模板
            prompt = ChatPromptTemplate.from_messages([
                ("system", """你是一个技术博客的智能助手。请基于提供的上下文准确回答用户问题。
                如果上下文不足，请诚实说明。回答要清晰、准确、有条理。"""),
                ("user", """上下文信息：
                {context}

                问题：{question}

                请基于上述上下文回答问题，如果上下文不相关或不足，请说明情况。""")
            ])
            
            # 生成回答
            chain = prompt | self.llm
            response = chain.invoke({
                "context": context,
                "question": query
            })
            
            answer = response.content
            print(f"✅ [{mode.upper()}模式] 回答生成完成")
            
            return {
                "answer": answer,
                "next_step": "validate_answer"
            }
        except Exception as e:
            error_msg = f"生成回答时出错: {str(e)}"
            print(f"❌ {error_msg}")
            return {
                "answer": f"抱歉，生成回答时遇到问题：{str(e)}",
                "next_step": "validate_answer"
            }
    
    def _validate_answer(self, state: GraphState) -> GraphState:
        """验证回答阶段"""
        answer = state["answer"]
        mode = state.get("mode", "io")
        
        # 简单的质量检查
        if len(answer.strip()) < 10:
            validated_answer = "回答内容过短，可能需要更多上下文信息。"
        elif "抱歉" in answer and "上下文" in answer:
            validated_answer = answer
        else:
            validated_answer = answer
        
        print(f"✅ [{mode.upper()}模式] 回答验证完成")
        
        return {
            "answer": validated_answer,
            "next_step": "end"
        }
    
    async def ask_question(self, question: str, mode: str = "io") -> str:
        """对外提供的问答接口"""
        print(f"\n{'='*60}")
        print(f"📝 用户问题: {question}")
        print(f"🔄 模式: {mode.upper()}")
        print(f"{'='*60}")
        
        # 初始化状态
        initial_state = {
            "messages": [HumanMessage(content=question)],
            "context": "",
            "answer": "",
            "next_step": "retrieve_context",
            "mode": mode,
            "streaming_results": [],
            "error": ""
        }
        
        # 执行图流程
        final_state = await self.graph.ainvoke(initial_state)
        
        return final_state["answer"]

3. 主程序

# client/main.py
import asyncio
from mcp_client import MCPClient
from graph_builder import BlogQAGraph
import os

# 从环境变量获取 OpenAI API Key
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("请设置 OPENAI_API_KEY 环境变量")

async def demo_both_modes():
    """演示两种模式"""
    
    questions = [
        "什么是 Python 装饰器？",
        "如何优化数据库查询性能？"
    ]
    
    modes = ["io", "sse"]
    
    async with MCPClient() as mcp_client:
        # 使用 gpt-3.5-turbo 模型
        qa_graph = BlogQAGraph(mcp_client)
        
        for question in questions:
            for mode in modes:
                print(f"\n🚀 处理问题: '{question}' (模式: {mode})")
                
                # 获取回答
                answer = await qa_graph.ask_question(question, mode=mode)
                
                # 输出结果
                print(f"\n💡 系统回答 ({mode.upper()}模式):")
                print("-" * 40)
                print(answer)
                print("-" * 40)
                
                # 添加延迟避免请求过快
                await asyncio.sleep(1)

async def interactive_mode():
    """交互式模式"""
    async with MCPClient() as mcp_client:
        qa_graph = BlogQAGraph(mcp_client)
        
        print("🎯 欢迎使用技术博客智能问答系统！")
        print("支持两种模式:")
        print("  io  - IO模式 (等待完整结果)")
        print("  sse - SSE模式 (流式接收结果)")
        print("输入 'quit' 或 'exit' 退出程序")
        print("-" * 50)
        
        while True:
            try:
                question = input("\n请输入您的技术问题: ").strip()
                
                if question.lower() in ['quit', 'exit', '退出']:
                    print("👋 再见！")
                    break
                
                if not question:
                    print("请输入有效问题")
                    continue
                
                # 选择模式
                mode = input("选择模式 (io/sse, 默认io): ").strip().lower()
                if mode not in ['io', 'sse']:
                    mode = 'io'
                
                answer = await qa_graph.ask_question(question, mode=mode)
                print(f"\n💡 回答 ({mode.upper()}模式):")
                print("-" * 40)
                print(answer)
                print("-" * 40)
                
            except KeyboardInterrupt:
                print("\n\n👋 程序已退出")
                break
            except Exception as e:
                print(f"❌ 发生错误: {e}")

if __name__ == "__main__":
    print("选择运行模式:")
    print("1. 演示模式 (两种模式对比)")
    print("2. 交互模式 (手动输入)")
    
    choice = input("请选择 (1/2): ").strip()
    
    if choice == "1":
        asyncio.run(demo_both_modes())
    elif choice == "2":
        asyncio.run(interactive_mode())
    else:
        print("无效选择，运行演示模式...")
        asyncio.run(demo_both_modes())

步骤三：依赖文件

# requirements.txt
flask==2.3.3
aiohttp==3.8.5
langgraph==0.0.15
langchain==0.0.352
langchain-openai==0.0.5
openai==1.3.7
python-dotenv==1.0.0

步骤四：运行说明

1. 启动服务端

# 在 server 目录下
python mcp_server.py

输出：

🚀 MCP Server 启动中...
服务地址: http://localhost:5000
同步接口: POST /api/search
流式接口: POST /api/search/stream

2. 运行客户端

# 设置环境变量
export OPENAI_API_KEY=your-openai-api-key

# 在 client 目录下
python main.py

3. 选择演示模式

选择 “1” 查看两种模式对比：

============================================================
📝 用户问题: 什么是 Python 装饰器？
🔄 模式: IO
============================================================
🔍 [IO模式] 正在检索关于 '什么是 Python 装饰器？' 的上下文...
📚 [IO模式] 检索到 2 条相关上下文
🤖 [IO模式] 正在生成回答...
✅ [IO模式] 回答生成完成
✅ [IO模式] 回答验证完成

💡 系统回答 (IO模式):
----------------------------------------
Python 装饰器是一种特殊类型的函数...

============================================================
📝 用户问题: 什么是 Python 装饰器？
🔄 模式: SSE
============================================================
🔍 [SSE模式] 正在检索关于 '什么是 Python 装饰器？' 的上下文...
📥 [SSE模式] 接收到结果 1
📥 [SSE模式] 接收到结果 2
🏁 [SSE模式] 检索完成，共 2 条结果
🤖 [SSE模式] 正在生成回答...
✅ [SSE模式] 回答生成完成
✅ [SSE模式] 回答验证完成

💡 系统回答 (SSE模式):
----------------------------------------
Python 装饰器是一种特殊类型的函数...

两种模式的区别

IO 模式特点：

• 同步等待：客户端发送请求后等待服务器返回完整结果
• 简单直接：实现简单，适合结果较小的场景
• 延迟较高：需要等待所有结果处理完成

SSE 模式特点：

• 流式传输：服务器逐步返回结果，客户端实时接收
• 用户体验好：用户可以逐步看到结果，减少等待感
• 资源利用高：适合处理大量数据或长时间运行的任务

这个完整的实现提供了准确的服务端和客户端代码，支持两种不同的通信模式，可以直接运行和测试。