停止为你的AI智能体编写自定义API

可将此内容视为《别再为你的 AI 智能体编写自定义 API：5 分钟搭建 MCP 服务器》一文中面向操作人员的思路重构版本：清晰的阶段划分、有序的代码模块，以及便于交接时参考的恢复说明。将概览视为可度量的框架使用效果最佳，在扩大范围之前，先记录一份理想的操作流程、一个故障案例以及回滚说明。相比冗长的脚本，应优先选择小型且可测试的单元。当某一步骤出错时，故障应能指向单一责任点，而非复杂的流程链。

步骤 1：架构与前置条件

对于第一步“架构与前提条件”，在修改代码之前需明确输入内容、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。

pip install mcp

第二步：构建MCP服务器

对于第二步：构建MCP服务器，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境切换到共享环境时出现意外费用。

import sqlite3
import json
import os
import sys
from mcp.server.mcpserver import MCPServer

# Initialize the MCP server
mcp = MCPServer(name="Enterprise_SQL_Agent")

# Force the database to be created in the exact same folder as this script
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "enterprise.db")

def setup_dummy_db():
    """Create a sample employee database for the demo"""
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()

        cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                          (id INTEGER PRIMARY KEY, name TEXT, role TEXT, salary INTEGER)''')
        cursor.execute("DELETE FROM employees")

        employees = [
            ("Alice", "Data Scientist", 120000),
            ("Bob", "DevOps Engineer", 115000),
            ("Charlie", "AI Researcher", 135000)
        ]

        cursor.executemany("INSERT INTO employees (name, role, salary) VALUES (?, ?, ?)", employees)
        conn.commit()
        conn.close()
        print("Database initialized successfully.", file=sys.stderr)
    except Exception as e:
        print(f"Database setup error: {e}", file=sys.stderr)

第三步：将数据库暴露给AI

对于第3步：将数据库暴露给AI，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。 对于第3步：将数据库暴露给AI，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。

@mcp.tool()
def query_employee_database(sql_query: str) -> str:
    """
    Executes a SQL SELECT query against the enterprise.db database.

    The database contains an 'employees' table with columns:
    - id (INTEGER PRIMARY KEY)
    - name (TEXT)
    - role (TEXT)
    - salary (INTEGER)

    SECURITY: Only READ operations (SELECT) are permitted.
    """

    # Safety Check: Block destructive SQL commands
    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
    if any(keyword in sql_query.upper() for keyword in dangerous_keywords):
        return "Error: Only SELECT queries are authorized for this tool."

    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute(sql_query)
        results = cursor.fetchall()

        # Format the output as JSON so the LLM can read it cleanly
        column_names = [description[0] for description in cursor.description]
        formatted_results = [dict(zip(column_names, row)) for row in results]

        conn.close()
        return json.dumps(formatted_results, indent=2)

    except Exception as e:
        return f"Database error: {str(e)}"

if __name__ == "__main__":
    setup_dummy_db()
    mcp.run()

第4步：连接 Claude Desktop

在执行第4步“连接 Claude Desktop”时，首先需列出相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关组件命名，明确成功判定标准，杜绝无声的半完成状态。 需记录每次调用的工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试代理循环将耗费大量时间。

{
  "mcpServers": {
    "enterprise-sql": {
      "command": "C:\\Users\\YourName\\.conda\\envs\\your_env\\python.exe",
      "args": [
        "D:\\Your\\Project\\Path\\mcp_server.py"
      ]
    }
  }
}

第5步：收获成果

在完成第5步“收益分析”时，首先写下合同细节：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及代币或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。没有这些记录，调试循环会浪费大量时间。

接下来做什么？

在规划下一步工作时，首先需明确接口规范：所需的输入参数、成功标识以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，以便操作人员无需查看整个系统结构即可进行审计。 需为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。若没有这些记录，调试过程将会浪费大量时间。 在规划下一步工作时，首先需明确接口规范：所需的输入参数、成功标识以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤出现故障时，故障原因应能明确指向某个具体的功能模块，而非整个复杂的流程。

运营检查清单

在编写操作检查清单时，首先记下合同要求：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。

同时记录正常流程和恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续的完善工作。

为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。没有这些记录，调试代理将陷入无休止的循环，浪费大量时间。

保持图结构简洁且具有类型约束。嵌套的数据结构会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续执行。

只要预算允许，就在持续集成环境中使用测试用例而非真实的付费 API 来添加能够检测关键路径的冒烟测试。

在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。

在推广该技术栈之前，应先冻结版本，为关键路径生成标准记录，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及负责密钥轮换的明确责任人。与其展示花哨的一次性演示，不如注重扎实的可靠性。

aed9f8a61db3的批处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌使用上限，并将记录存储在评估用示例文件旁边，以便后续模型更换时保持数据可比性。