uv add fastmcp

from fastmcp import FastMCP

mcp = FastMCP("MyServer")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    # Start an HTTP server on port 8000
    mcp.run(transport="http", host="127.0.0.1", port=8000)
uv run concepts-mcp/mcp_server_sample.py

╭────────────────────────────────────────────────────────────────────────────╮
│                                                                            │
│        _ __ ___  _____           __  __  _____________    ____    ____     │
│       _ __ ___ .'____/___ ______/ /_/  |/  / ____/ __ \  |___ \  / __ \    │
│      _ __ ___ / /_  / __ `/ ___/ __/ /|_/ / /   / /_/ /  ___/ / / / / /    │
│     _ __ ___ / __/ / /_/ (__  ) /_/ /  / / /___/ ____/  /  __/_/ /_/ /     │
│    _ __ ___ /_/    \____/____/\__/_/  /_/\____/_/      /_____(*)____/      │
│                                                                            │
│                                                                            │
│                                FastMCP  2.0                                │
│                                                                            │
│                                                                            │
│               ???  Server name:     MyServer                                 │
│               ?? Transport:       Streamable-HTTP                          │
│               ?? Server URL:      http://127.0.0.1:8000/mcp                │
│                                                                            │
│               ???  FastMCP version: 2.12.0                                   │
│               ?? MCP SDK version: 1.13.1                                   │
│                                                                            │
│               ?? Docs:            https://gofastmcp.com                    │
│               ?? Deploy:          https://fastmcp.cloud                    │
│                                                                            │
╰────────────────────────────────────────────────────────────────────────────╯


[09/01/25 22:54:03] INFO     Starting MCP server 'MyServer' with transport 'http' on                      server.py:1571
                             http://127.0.0.1:8000/mcp
INFO:     Started server process [24590]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

npm install -g @modelcontextprotocol/inspector

npx @modelcontextprotocol/inspector

npx @modelcontextprotocol/inspector
Starting MCP inspector...
?? Proxy server listening on 127.0.0.1:6277
?? Session token: ab9f8a6edafccaa8a276b5bfebfc1d4t0fe486b14b5b42208d11d777cd7f17b4
Use this token to authenticate requests or set DANGEROUSLY_OMIT_AUTH=true to disable auth

?? Open inspector with token pre-filled:
   http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=ab9f8a6edafccaa8a276b5bfebfc1d4t0fe486b14b5b42208d11d777cd7f17b4

?? MCP Inspector is up and running at http://127.0.0.1:6274 ??

uv add langchain-mcp-adapters
uv add langchain langchain-openai langchain-deepseek langgraph python-dotenv

DEEPSEEK_API_KEY=sk-……

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

import os
from dotenv import load_dotenv

# 加載.env文件中的環境變量
load_dotenv()

# Initialize the model
model = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

# Set up MCP client
client = MultiServerMCPClient(
    {
        "math": {
            # make sure you start your math server on port 8000
            "url": "http://127.0.0.1:8000/mcp/",
            "transport": "streamable_http",
        }
    }
)

asyncdef main():
    # Get tools from MCP server
    print("\n=== 獲取MCP工具 ===")
    tools = await client.get_tools()
    print(f"可用工具: {[tool.name for tool in tools]}")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    # Bind tools to model
    print("\n=== 綁定工具到模型 ===")
    model_with_tools = model.bind_tools(tools)
    print(f"已將 {len(tools)} 個工具綁定到模型")
    
    # Create ToolNode
    tool_node = ToolNode(tools)
    
    def should_continue(state: MessagesState):
        messages = state["messages"]
        last_message = messages[-1]
        if last_message.tool_calls:
            return"tools"
        return END
    
    # Define call_model function
    asyncdef call_model(state: MessagesState):
        messages = state["messages"]
        print("\n=== 調用LLM模型 ===")
        print(f"輸入消息數量: {len(messages)}")
        if messages:
            print(f"最新消息: {messages[-1].content if hasattr(messages[-1], 'content') else str(messages[-1])}")
        
        response = await model_with_tools.ainvoke(messages)
        print(f"模型響應類型: {type(response).__name__}")
        if hasattr(response, 'content'):
            print(f"響應內容: {response.content}")
        if hasattr(response, 'tool_calls') and response.tool_calls:
            print(f"工具調用: {len(response.tool_calls)} 個")
            for i, tool_call in enumerate(response.tool_calls):
                print(f"  工具 {i+1}: {tool_call['name']} - 參數: {tool_call['args']}")
        
        return {"messages": [response]}
    
    # Build the graph
    print("\n=== 構建LangGraph工作流 ===")
    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_node("tools", tool_node)
    print("已添加節點: call_model (模型調用) 和 tools (工具執行)")
    
    builder.add_edge(START, "call_model")
    builder.add_conditional_edges(
        "call_model",
        should_continue,
    )
    builder.add_edge("tools", "call_model")
    
    # Compile the graph
    graph = builder.compile()
    
    # Test the graph
    print("\n=== 開始測試數學計算 ===")
    test_question = "what's (3 + 5) x 12?"
    print(f"測試問題: {test_question}")
    
    math_response = await graph.ainvoke(
        {"messages": [{"role": "user", "content": test_question}]}
    )
    
    print("\n=== 最終結果 ===")
    print(f"消息鏈長度: {len(math_response['messages'])}")
    for i, msg in enumerate(math_response['messages']):
        msg_type = type(msg).__name__
        if hasattr(msg, 'content'):
            print(f"消息 {i+1} ({msg_type}): {msg.content}")
        else:
            print(f"消息 {i+1} ({msg_type}): {str(msg)}")

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

=== 獲取MCP工具 ===
可用工具: ['add', 'multiply']
  - add: Add two numbers
  - multiply: Multiply two numbers

=== 綁定工具到模型 ===
已將 2 個工具綁定到模型

=== 構建LangGraph工作流 ===
已添加節點: call_model (模型調用) 和 tools (工具執行)

=== 開始測試數學計算 ===
測試問題: what's (3 + 5) x 12?

=== 調用LLM模型 ===
輸入消息數量: 1
最新消息: what's (3 + 5) x 12?
模型響應類型: AIMessage
響應內容: I'll help you calculate (3 + 5) × 12. Let me break this down step by step.
工具調用: 1 個
  工具 1: add - 參數: {'a': 3, 'b': 5}

=== 調用LLM模型 ===
輸入消息數量: 3
最新消息: 8
模型響應類型: AIMessage
響應內容: Now I'll multiply the result (8) by 12:
工具調用: 1 個
  工具 1: multiply - 參數: {'a': 8, 'b': 12}

=== 調用LLM模型 ===
輸入消息數量: 5
最新消息: 96
模型響應類型: AIMessage
響應內容: The result of (3 + 5) × 12 is **96**. 

Here's the calculation:
- First, 3 + 5 = 8
- Then, 8 × 12 = 96

=== 最終結果 ===
消息鏈長度: 6
消息 1 (HumanMessage): what's (3 + 5) x 12?
消息 2 (AIMessage): I'll help you calculate (3 + 5) × 12. Let me break this down step by step.
消息 3 (ToolMessage): 8
消息 4 (AIMessage): Now I'll multiply the result (8) by 12:
消息 5 (ToolMessage): 96
消息 6 (AIMessage): The result of (3 + 5) × 12 is **96**. 

Here's the calculation:
- First, 3 + 5 = 8
- Then, 8 × 12 = 96

#!/usr/bin/env python3
"""
簡化的 FastMCP + LangGraph 演示
展示如何在FastMCP中集成LangGraph工作流
"""

import asyncio
import os
from typing import TypedDict, List
from datetime import datetime

from fastmcp import FastMCP, Context
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate

# 加載環境變量
from dotenv import load_dotenv
load_dotenv()

# 定義簡單的狀態類型
class TextProcessState(TypedDict):
    input_text: str
    processed_text: str
    ai_response: str
    steps: List[str]

# 初始化DeepSeek模型
model = ChatDeepSeek(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7
)

# 創建FastMCP實例
mcp = FastMCP("Simple-FastMCP-LangGraph")

def create_text_processing_graph():
    """創建文本處理的LangGraph工作流"""
    
    asyncdef preprocess_text(state: TextProcessState) -> TextProcessState:
        """預處理文本"""
        input_text = state["input_text"]
        
        # 簡單的預處理：去除多余空格，添加時間戳
        processed = input_text.strip()
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        return {
            **state,
            "processed_text": processed,
            "steps": state["steps"] + [f"文本預處理完成 ({timestamp})"]
        }
    
    asyncdef generate_ai_response(state: TextProcessState) -> TextProcessState:
        """生成AI響應"""
        processed_text = state["processed_text"]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一個專業的文本分析助手。請對用戶提供的文本進行分析和總結，提供有價值的見解。"),
            ("human", "請分析以下文本：\n\n{text}\n\n請提供：1) 主要內容總結 2) 關鍵信息提取 3) 簡短評價")
        ])
        
        try:
            response = await model.ainvoke(prompt.format_messages(text=processed_text))
            ai_content = response.content
        except Exception as e:
            ai_content = f"AI處理出錯: {str(e)}"
        
        return {
            **state,
            "ai_response": ai_content,
            "steps": state["steps"] + ["AI分析完成"]
        }
    
    # 構建工作流圖
    workflow = StateGraph(TextProcessState)
    
    # 添加節點
    workflow.add_node("preprocess", preprocess_text)
    workflow.add_node("ai_analyze", generate_ai_response)
    
    # 添加邊
    workflow.add_edge(START, "preprocess")
    workflow.add_edge("preprocess", "ai_analyze")
    workflow.add_edge("ai_analyze", END)
    
    # 編譯圖
    memory = MemorySaver()
    return workflow.compile(checkpointer=memory)

# 創建全局的LangGraph實例
text_processor = create_text_processing_graph()

@mcp.tool()
asyncdef process_text_with_langgraph(text: str, ctx: Context = None) -> str:
    """
    使用LangGraph處理文本
    
    Args:
        text: 要處理的文本內容
    
    Returns:
        處理結果
    """
    returnawait _analyze_text(text, ctx)

# 原始工具函數（不使用裝飾器）
asyncdef _analyze_text(text: str, ctx = None) -> str:
    """內部文本分析函數"""
    if ctx:
        await ctx.info(f"開始分析文本: {text[:30]}...")
    
    try:
        # 初始狀態
        initial_state = {
            "input_text": text,
            "processed_text": "",
            "ai_response": "",
            "steps": []
        }
        
        # 配置
        config = {
            "configurable": {
                "thread_id": f"analyze_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
            }
        }
        
        if ctx:
            await ctx.info("執行LangGraph工作流...")
        
        # 運行工作流
        final_state = await text_processor.ainvoke(initial_state, config)
        
        if ctx:
            await ctx.info("分析完成")
        
        # 格式化結果
        result = f"""?? 文本分析結果

?? 原始文本:
{final_state['input_text']}

?? AI分析:
{final_state['ai_response']}

?? 處理步驟:
{' → '.join(final_state['steps'])}

? 完成時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
        
        return result
        
    except Exception as e:
        error_msg = f"文本分析失敗: {str(e)}"
        if ctx:
            await ctx.error(error_msg)
        return error_msg

if __name__ == "__main__":
    import sys
    
    # 服務器模式
    print("?? 啟動 Simple FastMCP + LangGraph 服務器")
    print("?? 可用工具:")
    print("   - process_text_with_langgraph: 使用LangGraph處理文本")
    print("?? 服務器地址: http://127.0.0.1:8004/mcp")
    print("?? 測試命令: python simple_fastmcp_demo.py --test")
    print("=" * 60)
    
    mcp.run(
        transport="http",
        host="127.0.0.1",
        port=8004,
        log_level="info"
    )

啟動與驗證
在啟動前，請確保已在項目根目錄配置 .env 并設置 DEEPSEEK_API_KEY。

運行代碼
```Shell
uv run langgraph_use_mcp_as_server.py