# Windows安装 powershell -c "irm https://astral.sh/uv/install.ps1 | iex" # 项目初始化 uv init mcp_server cd mcp_server uv venv .\.venv\Scripts\activate
# 核心依赖安装 uv add mcp[cli] httpx psycopg2
在 https://nodejs.org/zh-cn/download，选择 v22.17.0（LTS） 版本进行下载并安装：
# 验证安装 node --version npm --version
1. 资源(Resource)开发
数据库连接模板：
DB_CONFIG = {
"dbname": "production_db",
"user": "admin",
"password": "secure_pass",
"host": "10.1.1.27",
"port": "11003"
}
def get_db_connection():
return psycopg2.connect(**DB_CONFIG)四类关键资源：
# 1. 基础测试资源
@mcp.resource("test://hello")
def hello() -> str:
return "Hello, MCP World!"
# 2. 表名查询
@mcp.resource("db://tables")
def list_tables() -> str:
with get_db_connection() as conn:
with conn.cursor() as cur:
cur.execute("SELECT table_name FROM information_schema.tables")
return json.dumps([row[0] for row in cur.fetchall()])
# 3. 表数据查询（防SQL注入）
@mcp.resource("db://tables/{table_name}/data")
def get_table_data(table_name: str, limit: int = 100) -> str:
with get_db_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT * FROM %s LIMIT %s",
(psycopg2.extensions.AsIs(table_name), limit))
return json.dumps(cur.fetchall(), default=str)
# 4. 表结构查询
@mcp.resource("db://tables/{table_name}/schema")
def get_table_schema(table_name: str) -> str:
with get_db_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = %s
""", (table_name,))
return json.dumps([dict(name=row[0], type=row[1]) for row in cur.fetchall()])2. 提示(Prompt)开发
省级介绍模板：
@mcp.prompt()
def introduce_province(province: str) -> str:
return f"""
请从以下维度介绍{province}：
1. 历史沿革
2. 人文地理特点
3. 经济发展现状
4. 特色旅游资源
"""代码调试模板（多轮对话）：
@mcp.prompt()
def debug_code(code: str, error: str) -> list[base.Message]:
return [
base.SystemMessage("你是一位专业的代码调试助手"),
base.UserMessage("请帮我修复以下代码："),
base.UserMessage(f"```python\n{code}\n```"),
base.UserMessage(f"错误信息：\n{error}"),
base.AssistantMessage("我将按以下步骤分析：1. 语法检查 2. 逻辑分析 3. 修复方案")
]3. 工具(Tool)开发
数学运算工具集：
@mcp.tool()
def add(a: float, b: float) -> float:
"""加法运算 (a + b)"""
return a + b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""除法运算 (a / b)"""
if b == 0:
raise ValueError("除数不能为零")
return a / b服务启动配置：
if __name__ == "__main__":
mcp = FastMCP(
"Production MCP Server",
debug=False, # 生产环境关闭调试
host="0.0.0.0",
port=8000
)
mcp.run('sse') # 生产环境使用SSE协议在终端运行 mcp --help，可以查看 mcp 命令的用法（由下面的返回结果可知，可以通过 mcp dev 命令运行 mcp inspector）：
>mcp --help
mcp dev 命令语法：
mcp dev --help
# 错误方式（漏洞）
cur.execute(f"SELECT * FROM {table_name}")
# 正确方式（参数化）
cur.execute("SELECT * FROM %s", (psycopg2.extensions.AsIs(table_name),))# 生产环境限制访问IP mcp = FastMCP(host="192.168.1.100")
@mcp.resource("db://secure/data")
def secure_data():
try:
# 业务逻辑
except Exception as e:
return json.dumps({"error": str(e)})性能优化方案：
# 1. 连接池管理
from psycopg2.pool import SimpleConnectionPool
pool = SimpleConnectionPool(5, 20, **DB_CONFIG)
# 2. 异步处理
@mcp.resource("db://async")
async def async_data():
async with pool.getconn() as conn:
# 异步查询通过Resource/Prompt/Tool三层抽象，MCP实现了：
安全隔离：工具操作通过沙箱环境执行
标准化接口：统一AI模型与外部系统交互规范
灵活扩展：支持数据库/API/文件系统等多类连接
开发效率：Inspector工具实现可视化调试
本文相关MCP实战代码联系官网客服
