使用说明

SQL Server MCP服务器

一种模型上下文协议（MCP）服务器，提供与SQL server数据库交互的工具。此服务器允许您通过MCP兼容客户端从SQL server数据库中查询、添加和删除人员记录。

特性

查询人员：按姓名在数据库中搜索人员或检索所有记录
添加人员：插入包含姓名和出生日期的新人员记录
删除个人：根据姓名和出生日期删除人员记录
优化连接管理：使用FastMCP的生命周期管理实现高效的数据库连接
类型安全：完全使用Pydantic模型和适当的错误处理

先决条件

数据库要求

SQL Server实例（本地或远程）
带有a的数据库 people 桌子
已安装SQL Server的ODBC驱动程序18

必需的表架构

您的SQL Server数据库应具有 people 表具有以下结构：

CREATE TABLE people (
    id INT IDENTITY(1,1) PRIMARY KEY,
    name NVARCHAR(100) NOT NULL,
    surname NVARCHAR(100) NOT NULL,
    birthdate DATE NOT NULL
);

ODBC驱动程序安装

从以下位置下载并安装适用于SQL Server的ODBC驱动程序18： https://learn.microsoft.com/en-us/sql/connect/odbc/download-odbc-driver-for-sql-server

安装

克隆或下载此存储库
使用uv（推荐）或pip安装依赖项：

# Using uv (recommended)
uv sync

# Or using pip
pip install -r requirements.txt

配置

创建一个 .env 项目根目录中的文件，其中包含您的SQL Server连接详细信息：

MSSQL_SERVER=your-server-name-or-ip
MSSQL_DATABASE=your-database-name
MSSQL_USER=your-username
MSSQL_PASSWORD=your-password

示例 `.env` 文件：

MSSQL_SERVER=localhost
MSSQL_DATABASE=TestDB
MSSQL_USER=sa
MSSQL_PASSWORD=YourPassword123!

用法

运行服务器

python server.py

服务器将启动并连接到您的SQL server数据库。它使用FastMCP的生命周期管理在服务器的整个生命周期内维护持久的数据库连接。

可用工具

1.查询人员

在数据库中搜索人员：

无过滤器：返回数据库中的所有人
带名称过滤器：返回与指定名称匹配的人员

2.添加人员

使用以下命令将新用户添加到数据库中：

姓名（必填）
姓氏（必填）
出生日期（必填，YYYY-MM-DD格式）

3.删除人员

通过提供以下信息从数据库中删除人员：

姓名（必填）
姓氏（必填）
出生日期（必填，YYYY-MM-DD格式）

备注：如果多条记录符合条件，则所有记录都将被删除。

工作原理：技术深度潜水

本节解释Python新手的技术概念和架构。

理解代码结构

1.进口和依赖关系

import os
import pyodbc
from datetime import date
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from fastmcp import FastMCP, Context
from dotenv import load_dotenv
from pydantic import BaseModel, Field

每次导入的作用：

os：提供对环境变量（如数据库凭据）的访问
pyodbc：用于连接SQL Server数据库的Python库
datetime.date：用于处理日期对象（出生日期）
AsyncIterator：异步迭代器的类型提示（Python高级功能）
asynccontextmanager：用于创建异步上下文管理器的装饰器（如下所述）
dataclass：用于创建简单数据类的装饰器
FastMCP, Context：核心MCP服务器框架组件
load_dotenv：从加载环境变量 .env 文件
BaseModel, Field：用于数据验证的Pydantic类

2.加载环境变量

load_dotenv()

这行读你的 .env 文件，并通过以下方式提供数据库凭据 os.getenv()。这是一种安全最佳实践-永远不要在源代码中硬编码密码！

3.数据库连接类

class DatabaseConnection:
    """Database connection manager for SQL Server."""
    
    def __init__(self):
        self.connection = None
    
    async def connect(self) -> "DatabaseConnection":
        """Connect to SQL Server database."""
        self.connection = pyodbc.connect(
            f"DRIVER={{ODBC Driver 18 for SQL Server}};"
            f"SERVER={os.getenv('MSSQL_SERVER')};"
            f"DATABASE={os.getenv('MSSQL_DATABASE')};"
            f"UID={os.getenv('MSSQL_USER')};"
            f"PWD={os.getenv('MSSQL_PASSWORD')};"
            "Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;"
        )
        return self

理解这个类：

__init__：将连接初始化为的构造函数 None
async def connect(): 异步函数 建立数据库连接
为什么异步？ 数据库连接可能需要时间，异步允许其他操作在等待时继续进行
连接字符串：告诉pyodbc如何使用凭据连接到SQL Server
安全设置: Encrypt=yes 确保数据在传输过程中加密

4.应用程序上下文（依赖注入）

@dataclass
class AppContext:
    """Application context with typed dependencies."""
    db: DatabaseConnection

这是什么？

A. 数据类 是创建仅保存数据的类的简单方法
这将存储我们的数据库连接，以便所有工具都可以访问它
依赖注入：不是每个工具都创建自己的连接，而是共享这个连接

5.寿命管理（效率背后的魔力）

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context."""
    # Initialize on startup
    db = await DatabaseConnection().connect()
    try:
        yield AppContext(db=db)
    finally:
        # Cleanup on shutdown
        await db.disconnect()

了解异步上下文管理器：

发生了什么：此函数在服务器启动和停止时运行
@asynccontextmanager：使此函数成为上下文管理器的装饰器
await DatabaseConnection().connect()：等待数据库连接建立
yield AppContext(db=db)：为所有工具提供数据库连接
try/finally：确保即使出现问题，也能进行数据库清理
为什么这很重要：整个服务器生命周期只有一个连接，而不是每次操作都连接/断开连接

6.服务器初始化

mcp = FastMCP("MSSQL MCP Server", lifespan=app_lifespan)

这创建了MCP服务器，并告诉它使用我们的寿命管理器进行数据库连接。

7.数据模型（类型安全）

class Person(BaseModel):
    name: str = Field(..., description="Name of the person")
    surname: str = Field(..., description="Surname of the person")
    birthdate: date = Field(..., description="Birthdate of the person")

理解Pydantic模型：

基本模型：提供自动数据验证
字段（…）：The ... 表示“必填字段”
键入提示: str 和 date 告诉Python需要什么类型的数据
验证：Pydantic会自动验证姓名是否为字符串，出生日期是否为有效日期

8.MCP工具（主要功能）

查询工具示例：

@mcp.tool()
def query_people(ctx: Context, name: str = None) -> list[Person]:
    """Query people from the database."""
    db = ctx.request_context.lifespan_context.db
    cursor = db.get_cursor()
    # ... database operations

分解如下：

@mcp.tool()：将此函数注册为MCP工具的装饰器
ctx: Context：上下文包含我们的数据库连接
ctx.request_context.lifespan_context.db：访问我们共享数据库连接的路径
cursor = db.get_cursor()：获取游标以执行SQL命令
返回类型 list[Person]：告诉Python这将返回Person对象的列表

连接流程图

Server Startup:
1. load_dotenv() → Load .env file
2. app_lifespan() → Create database connection
3. yield AppContext(db) → Make connection available

Tool Execution:
1. Client calls tool → FastMCP receives request
2. Tool accesses ctx.request_context.lifespan_context.db → Get shared connection
3. Execute SQL → Use cursor to run database commands
4. Return results → Send back to client

Server Shutdown:
1. finally block → Close database connection
2. Clean exit → All resources freed

为什么这种架构是高效的

传统方法（低效）：

Tool 1: Connect → Query → Disconnect
Tool 2: Connect → Insert → Disconnect  
Tool 3: Connect → Delete → Disconnect

我们的优化方法：

Startup: Connect once
Tool 1: Query (reuse connection)
Tool 2: Insert (reuse connection)
Tool 3: Delete (reuse connection)
Shutdown: Disconnect once

异步与同步：何时以及为什么

同步（阻塞）：

def slow_operation():
    time.sleep(5)  # Blocks everything for 5 seconds
    return "done"

异步（非阻塞）：

async def fast_operation():
    await asyncio.sleep(5)  # Other things can run during these 5 seconds
    return "done"

在我们的服务器中：

数据库连接使用 async 因为它们可能需要时间
服务器可以在一个请求等待数据库时处理多个请求
更好的性能和响应能力

错误处理和安全

SQL注入防护：

# ❌ DANGEROUS - Don't do this:
cursor.execute(f"SELECT * FROM people WHERE name = '{name}'")

# ✅ SAFE - Parameterized query:
cursor.execute("SELECT * FROM people WHERE name = ?", (name,))

安全环境变量：

数据库密码存储在 .env 文件（不在代码中）
.env 应添加到 .gitignore （切勿提交密码）
已启用连接加密(Encrypt=yes)

实例：理解代码流

示例1：逐步添加人员

当你打电话给 add_person 工具，具体情况如下：

@mcp.tool()
def add_person(ctx: Context, person: Person) -> str:
    # Step 1: Get the shared database connection
    db = ctx.request_context.lifespan_context.db
    
    # Step 2: Create a cursor for executing SQL
    cursor = db.get_cursor()
    
    # Step 3: Execute parameterized SQL (prevents SQL injection)
    cursor.execute(
        "INSERT INTO people (name, surname, birthdate) VALUES (?, ?, ?)",
        (person.name, person.surname, person.birthdate)
    )
    
    # Step 4: Commit the transaction (save changes)
    db.connection.commit()
    
    # Step 5: Clean up the cursor (but keep connection open)
    cursor.close()
    
    # Step 6: Return success message
    return f"Person {person.name} {person.surname} added successfully mate"

幕后发生了什么：

输入验证：Pydantic确保 person 具有有效的姓名和出生日期
连接复用：创建新的数据库连接不会浪费时间
安全SQL：The ? 占位符防止恶意SQL注入
交易: commit() 使更改永久化
资源管理：光标已关闭，但连接保持打开状态，以便进行下一次操作

示例2：理解上下文访问

路径 ctx.request_context.lifespan_context.db 可能看起来很复杂，但以下是每个部分的含义：

ctx                          # The context object passed to every tool
  .request_context          # Information about this specific request
    .lifespan_context       # Data that lives for the entire server lifetime
      .db                   # Our database connection object

把它想象成一个文件路径： ctx → request_context → lifespan_context → db

示例3：实践中的错误处理

以下是如何增强错误处理：

@mcp.tool()
def safe_add_person(ctx: Context, person: Person) -> str:
    try:
        db = ctx.request_context.lifespan_context.db
        cursor = db.get_cursor()
        
        cursor.execute(
            "INSERT INTO people (name, surname, birthdate) VALUES (?, ?, ?)",
            (person.name, person.surname, person.birthdate)
        )
        db.connection.commit()
        cursor.close()
        
        return f"Person {person.name} {person.surname} added successfully!"
        
    except pyodbc.IntegrityError as e:
        return f"Error: Duplicate person or invalid data - {str(e)}"
    except pyodbc.Error as e:
        return f"Database error: {str(e)}"
    except Exception as e:
        return f"Unexpected error: {str(e)}"

学习路径：掌握Python概念

如果你是Python新手，想更好地理解这段代码，请按顺序学习这些概念：

初级水平：

类和对象 -理解 class DatabaseConnection
功能与参数 -如何 def query_people(name: str = None) 作品
环境变量 -我们为什么使用 os.getenv('MSSQL_SERVER')
字符串格式化 -理解f字符串，如 f"Person {person.name}"

中级水平：

类型提示 -什么 -> list[Person] 手段
装饰器 -如何 @mcp.tool() 修改函数
上下文管理器 -理解 with 声明和 try/finally
异常处理 -使用 try/except 块

高级水平：

异步编程 -理解 async def 和 await
数据类 -如何 @dataclass 作品
Pydantic模型 -高级数据验证
依赖注入 -我们使用的上下文模式

数据库概念：

SQL基础知识 -SELECT、INSERT、DELETE语句
参数化查询 -防止SQL注入
数据库连接 -连接字符串和游标
交易 -何时使用 commit() 和 rollback()

依赖项

fastmcp：MCP服务器框架
pyodbc：SQL Server数据库连接
python dotenv：环境变量管理
媒染剂：数据验证和序列化

安全考虑

环境变量用于数据库凭据
通过参数化查询实现SQL注入保护
与SQL Server的加密连接（TrustServerCertificate=否）
可靠性连接超时设置

故障排除

常见问题

未找到ODBC驱动程序
- 确保已安装SQL Server的ODBC驱动程序18
- 检查连接字符串中的驱动程序名称是否与安装的版本匹配
连接失败
- 验证SQL Server是否正在运行且可访问
- 检查防火墙设置
- 在中验证凭据 .env 文件
- 确保数据库存在
找不到表
- 创建 people 具有所需架构的表
- 验证中的数据库名称 .env 文件
权限不足
- 确保数据库用户对具有SELECT、INSERT、DELETE权限 people 桌子
- 检查用户身份验证方法（SQL Server或Windows身份验证）

发展

要修改或扩展此服务器，请执行以下操作：

通过装饰功能添加新工具 @mcp.tool()
使用 ctx: Context 访问数据库连接的参数
通过以下方式访问数据库 ctx.request_context.lifespan_context.db
遵循现有的光标管理模式

许可证

这个项目是开源的。请检查许可证文件以了解详细信息。

贡献

分叉存储库
创建要素分支
进行更改
彻底测试
提交拉取请求

支持

对于问题和疑问：

检查上面的故障排除部分
查看SQL Server和MCP文档
在存储库中打开问题