AGI｜基于FastMCP 2.0的MCP Server快速搭建指南

Part1

介绍

Part2

MCP开发框架

Part3

快速搭建 MCP Server

# python 版本 要求 >= 3.10
# 使用 uv 来管理项目
# 初始化项目
uv init mcp-server-demo
cd mcp-server-demo

# 创建并激活环境
uv venv
source .venv/bin/activate

# 安装mcp 依赖
uv add "fastmap"

# server.py
# 引入 fastmcp 包
from fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("CalculatorServer")


# 定义工具，也就是实际执行的内容，类似于 RestAPI 中的 POST，mcp主要的能力都在这里体现。
# Add an addition tool
@mcp.tool()
defadd(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


# 定义资源，资源为只读，类似于 RestAPI 中的 Get,可用于获取配置、版本等静态资源信息。
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
defget_greeting(name: str) -> str:
    """Get a personalized greeting"""
    returnf"Hello, {name}!"
    
    
if __name__ == "__main__":
    mcp.run(transport='stdio')

Part4

搭建一个多 MCP Server 组合的复杂应用

import asyncio
from fastmcp import FastMCP, Client

# --- 定义子服务器（商品推荐服务） ---
# 阿里西西商城
product_recommendation_mcp = FastMCP(name="商品推荐服务")


# 定义获取热门商品的工具函数
@product_recommendation_mcp.tool()
def get_popular_products():
    """List of popular products"""
    # mock data
    hot_products = ('[{"product_name":"iPhone 14 Pro Max", "price": 5999, "category": "Electronics"},'
                    '{"product_name":"小米SU7 Ultra", "price": 529999, "category": "Car"},'
                    '{"product_name":"红米K80", "price": 2399, "category": "Electronics"}]')

    return hot_products


# 定义为指定用户获取推荐商品的工具函数
@product_recommendation_mcp.tool()
def get_recommended_products(user_id):
    """Products recommended for users"""
    # mock data
    recommended_products = '{"product_name":"小米SU7 Ultra", "price": 529999, "category": "Car"}'
    return recommended_products


# --- 定义子服务器（支付服务） ---
pay_mcp = FastMCP(name="支付服务")


# 定义支付工具函数
@pay_mcp.tool()
def pay(product_name: str, amount: float):
    """Pay for a product"""
    # 这里可以是请求API或者是具体支付逻辑
    # 或者有没有可能未来的各种系统，都会实现一套 mcp 的接口，然后接收apikey等鉴权信息，类比于Restful API
    # 其实很多大厂已经这样做了
    returnf"恭喜您支付成功，商品名称：{product_name}，支付金额{amount}元。"


# --- 定义主服务器（电商主服务） ---
ecommerce_main_mcp = FastMCP(name="阿里西西商城服务")

# --- 挂载子服务器（同步操作） ---
ecommerce_main_mcp.mount("recommendation", product_recommendation_mcp)
ecommerce_main_mcp.mount("pay", pay_mcp)

print("已将子服务挂载到电商主服务上。")


# --- 在挂载后为子服务器添加一个工具 ---
# 定义获取新上架商品的工具函数
@product_recommendation_mcp.tool()
def get_new_arrivals():
    """List of newly listed products"""
    # mock data
    new_arrivals = ('[{"product_name":"新款华为Mate 60 Pro", "price": 6999, "category": "Electronics"},'
                    '{"product_name":"新款小米平板", "price": 2999, "category": "Electronics"}]')
    return new_arrivals


print("已为商品推荐子服务添加 'get_new_arrivals' 工具。")


# --- 测试访问 ---
asyncdef test_dynamic_mount():
    # 从主服务器获取所有可用的工具
    tools = await ecommerce_main_mcp.get_tools()
    print("通过电商主服务可用的工具:", list(tools.keys()))

    asyncwith Client(ecommerce_main_mcp) as client:
        # 通过主服务器调用 'get_popular_products' 工具
        popular_products_result = await client.call_tool("recommendation_get_popular_products")
        print("调用 recommendation_get_popular_products 的结果:", popular_products_result[0].text)

        # 通过主服务器调用 'get_recommended_products' 工具
        recommended_products_result = await client.call_tool("recommendation_get_recommended_products",
                                                             parameters={"user_id": "123"})
        print("调用 recommendation_get_recommended_products 的结果:", recommended_products_result[0].text)

        # 通过主服务器调用动态添加的 'get_new_arrivals' 工具
        new_arrivals_result = await client.call_tool("recommendation_get_new_arrivals")
        print("调用 recommendation_get_new_arrivals 的结果:", new_arrivals_result[0].text)


if __name__ == "__main__":
    # 需要异步上下文进行测试
    # asyncio.run(test_dynamic_mount())
    # 若要运行服务器本身:
    ecommerce_main_mcp.run()

########### 主程序 #################
from fastmcp import FastMCP
import asyncio

# Import the servers (see other files)
from modules.text_server import text_mcp
from modules.data_server import data_mcp

app = FastMCP(name="MainApplication")

# Setup function for async imports
asyncdef setup():
    # Import the utility servers
    await app.import_server("text", text_mcp)
    await app.import_server("data", data_mcp)

@app.tool()
def process_and_analyze(record_id: int) -> str:
    """Fetches a record and analyzes its string representation."""
    # In a real application, you'd use proper methods to interact between
    # imported tools rather than accessing internal managers
    
    # Get record data
    record = {"id": record_id, "value": random.random()}
    
    # Count words in the record string representation
    word_count = len(str(record).split())
    
    return (
        f"Record {record_id} has {word_count} words in its string "
        f"representation."
    )

if __name__ == "__main__":
    # Run async setup before starting the server
    asyncio.run(setup())
    # Run the server
    app.run()


########### 子程序-data_mcp #################
from fastmcp import FastMCP
import random
from typing import dict

data_mcp = FastMCP(name="DataAPI")

@data_mcp.tool()
def fetch_record(record_id: int) -> dict:
    """Fetches a dummy data record."""
    return {"id": record_id, "value": random.random()}

@data_mcp.resource("data://schema/{table}")
def get_table_schema(table: str) -> dict:
    """Provides a dummy schema for a table."""
    return {"table": table, "columns": ["id", "value"]}
    
########### 子程序-text_mcp #################
from fastmcp import FastMCP

text_mcp = FastMCP(name="TextUtilities")

@text_mcp.tool()
def count_words(text: str) -> int:
    """Counts words in a text."""
    return len(text.split())

@text_mcp.resource("resource://stopwords")
def get_stopwords() -> list[str]:
    """Return a list of common stopwords."""
    return ["the", "a", "is", "in"]

Part5

总结思考