 距离2026世界人工智能大会（WAIC 2026）开幕仅剩15天，上海世博中心正在紧锣密鼓地筹备这场被称为"AI界双十一"的年度盛会——超过300款AI新品将全球首发，涵盖大模型、具身智能、AI芯片等全产业链。与此同时，Anthropic于上周发布的Claude Sonnet 5再次引爆技术圈，其原生MCP（Model Context Protocol）深度集成能力，标志着AI智能体（Agent）生态正式从"概念验证"迈入"生产部署"阶段。

2026年，被业界公认为**"AI智能体元年"

。一个不争的事实是：AI不再只是"回答问题"的对话框，它正在学会调用工具、读取数据库、操作文件系统，甚至自主规划和修正任务——而这一切的核心基础设施，正是

在MCP出现之前，AI工具集成是一个碎片化的噩梦：

传统方式：
  Claude ──── 接口A ──── 数据库
  ChatGPT ──── 接口B ──── 数据库（重复适配）
  Gemini ──── 接口C ──── 数据库（再写一遍）

MCP方式：
  Claude ──┐
  ChatGPT ──┼── MCP协议 ──── 数据库（一次适配，处处可用）
  Gemini ──┘


MCP（Model Context Protocol）由Anthropic于2024年末发起，2025年OpenAI和Google相继宣布支持。三大竞争对手在技术标准上达成一致——这种事在互联网历史上每次发生，都意味着一个新时代的开始。

截至2026年6月，支持MCP的社区Server已超过

，覆盖数据库、文件操作、API网关、代码执行等几乎所有常见场景。

二、从零构建你的第一个MCP Server

理论讲够了，让我们动手。下面我们将构建一个

，它能让AI Agent自主查询数据库并生成分析报告。

# 创建项目
mkdir my-mcp-server && cd my-mcp-server
python -m venv venv && source venv/bin/activate

# 安装核心依赖
pip install mcp[cli] sqlalchemy


"""
mcp_db_server.py
一个生产级MCP Server：让AI Agent自主查询SQLite数据库
兼容 Claude Desktop / Cursor / 任何支持MCP的Host
"""

import sqlite3
import json
import os
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
)

# ============================================================
# 1. 初始化MCP Server
# ============================================================
server = Server("db-analyst-server")

# 数据库路径（可替换为PostgreSQL/MySQL连接串）
DB_PATH = os.getenv("DB_PATH", "company.db")


def get_db_connection():
    """获取数据库连接"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


# ============================================================
# 2. 定义工具（Tools）—— Agent可调用的能力
# ============================================================

@server.list_tools()
async def list_tools():
    """声明本Server提供的所有工具"""
    return [
        Tool(
            name="list_tables",
            description="列出数据库中所有的表名及其描述",
            inputSchema={
                "type": "object",
                "properties": {},
            },
        ),
        Tool(
            name="describe_table",
            description="获取指定表的完整结构信息（列名、类型、约束）",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {
                        "type": "string",
                        "description": "表名",
                    },
                },
                "required": ["table_name"],
            },
        ),
        Tool(
            name="execute_query",
            description="执行只读SQL查询并返回结果（仅支持SELECT语句）",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL SELECT语句",
                    },
                    "max_rows": {
                        "type": "integer",
                        "description": "最大返回行数，默认50",
                        "default": 50,
                    },
                },
                "required": ["sql"],
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict):
    """工具调用路由"""
    conn = get_db_connection()
    cursor = conn.cursor()

    try:
        if name == "list_tables":
            cursor.execute("""
                SELECT name FROM sqlite_master 
                WHERE type='table' AND name NOT LIKE 'sqlite_%'
            """)
            tables = [row[0] for row in cursor.fetchall()]
            result = {
                "database": DB_PATH,
                "tables": tables,
                "count": len(tables),
            }

        elif name == "describe_table":
            table = arguments["table_name"]
            cursor.execute(f"PRAGMA table_info('{table}')")
            columns = [
                {
                    "name": row[1],
                    "type": row[2],
                    "nullable": not row[3],
                    "default": row[4],
                    "primary_key": bool(row[5]),
                }
                for row in cursor.fetchall()
            ]
            # 获取行数
            cursor.execute(f'SELECT COUNT(*) FROM "{table}"')
            row_count = cursor.fetchone()[0]

            result = {
                "table": table,
                "columns": columns,
                "row_count": row_count,
            }

        elif name == "execute_query":
            sql = arguments["sql"].strip()
            max_rows = arguments.get("max_rows", 50)

            # 安全检查：仅允许SELECT
            if not sql.upper().startswith("SELECT"):
                return [TextContent(
                    type="text",
                    text=json.dumps({
                        "error": "安全限制：仅允许执行SELECT查询语句"
                    }, ensure_ascii=False),
                )]

            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchmany(max_rows)
            data = [dict(zip(columns, row)) for row in rows]

            result = {
                "sql": sql,
                "columns": columns,
                "row_count": len(data),
                "truncated": len(rows) >= max_rows,
                "data": data,
                "executed_at": datetime.now().isoformat(),
            }
        else:
            result = {"error": f"未知工具: {name}"}

        return [TextContent(
            type="text",
            text=json.dumps(result, ensure_ascii=False, default=str, indent=2),
        )]

    except Exception as e:
        return [TextContent(
            type="text",
            text=json.dumps({
                "error": str(e),
                "tool": name,
            }, ensure_ascii=False),
        )]
    finally:
        conn.close()


# ============================================================
# 3. 启动Server（stdio传输）
# ============================================================
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )


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


"""seed_db.py — 生成示例数据"""
import sqlite3
import random
from datetime import datetime, timedelta

conn = sqlite3.connect("company.db")
c = conn.cursor()

c.execute("""
    CREATE TABLE IF NOT EXISTS employees (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        department TEXT,
        salary REAL,
        hire_date TEXT
    )
""")

c.execute("""
    CREATE TABLE IF NOT EXISTS sales (
        id INTEGER PRIMARY KEY,
        employee_id INTEGER,
        product TEXT,
        amount REAL,
        sale_date TEXT,
        FOREIGN KEY (employee_id) REFERENCES employees(id)
    )
""")

departments = ["工程部", "市场部", "销售部", "产品部", "数据部"]
products = ["企业版License", "API调用包", "咨询服务", "培训课程"]

# 插入员工
for i in range(1, 51):
    c.execute(
        "INSERT INTO employees VALUES (?, ?, ?, ?, ?)",
        (
            i,
            f"员工{i:03d}",
            random.choice(departments),
            round(random.uniform(8000, 45000), 2),
            (datetime(2020, 1, 1) + timedelta(days=random.randint(0, 2000))).isoformat()[:10],
        ),
    )

# 插入销售记录
for i in range(1, 201):
    c.execute(
        "INSERT INTO sales VALUES (?, ?, ?, ?, ?)",
        (
            i,
            random.randint(1, 50),
            random.choice(products),
            round(random.uniform(1000, 100000), 2),
            (datetime(2025, 1, 1) + timedelta(days=random.randint(0, 540))).isoformat()[:10],
        ),
    )

conn.commit()
conn.close()
print("✅ 数据库初始化完成：50名员工，200条销售记录")


~/Library/Application Support/Claude/claude_desktop_config.json

重启Claude Desktop后，你只需对它说：

探查表结构 → 构造SQL → 执行查询 → 分析结果 → 生成报告

方式二：用Python构建自定义Agent Host

如果你想在自己的应用中集成MCP，以下是完整的Agent Host代码：

"""
agent_host.py — 自定义MCP Agent Host
使用Ollama本地模型 + MCP Server构建私有智能体
"""

import asyncio
import json
import httpx
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

OLLAMA_BASE = "http://localhost:11434"
MODEL = "qwen3:8b"  # 或其他支持function calling的模型


async def call_llm(messages: list, tools: list = None) -> dict:
    """调用Ollama本地大模型"""
    payload = {
        "model": MODEL,
        "messages": messages,
        "stream": False,
        "format": "json",
    }
    if tools:
        payload["tools"] = tools

    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.post(f"{OLLAMA_BASE}/api/chat", json=payload)
        return resp.json()["message"]


async def run_agent(user_query: str):
    """运行完整的Agent循环"""

    # 1. 连接MCP Server
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_db_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 2. 获取可用工具
            tools_resp = await session.list_tools()
            mcp_tools = tools_resp.tools

            # 将MCP工具转换为LLM可理解的格式
            llm_tools = []
            for t in mcp_tools:
                llm_tools.append({
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.inputSchema,
                    },
                })

            print(f"🔧 已加载 {len(mcp_tools)} 个工具: "
                  f"{[t.name for t in mcp_tools]}")

            # 3. Agent Loop
            messages = [
                {
                    "role": "system",
                    "content": (
                        "你是一个专业的数据分析助手。"
                        "你可以通过工具查询数据库来回答用户问题。"
                        "请遵循以下步骤：\n"
                        "1. 先用list_tables查看有哪些表\n"
                        "2. 用describe_table了解表结构\n"
                        "3. 构造SQL查询获取数据\n"
                        "4. 基于数据给出分析结论\n"
                        "每次只调用一个工具，根据返回结果决定下一步。"
                    ),
                },
                {"role": "user", "content": user_query},
            ]

            max_iterations = 10

            for step in range(max_iterations):
                print(f"\n{'='*50}")
                print(f"📍 Agent Step {step + 1}")
                print(f"{'='*50}")

                # 调用LLM
                response = await call_llm(messages, llm_tools)
                content = response.get("content", "")

                # 检查是否需要调用工具
                tool_calls = response.get("tool_calls", [])

                if not tool_calls:
                    # Agent给出最终回答
                    print(f"\n✅ 最终回答:\n{content}")
                    break otterly.cn

                # 执行工具调用
                for tc in tool_calls:
                    func_name = tc["function"]["name"]
                    func_args = json.loads(tc["function"]["arguments"])

                    print(f"🔧 调用工具: {func_name}")
                    print(f"   参数: {json.dumps(func_args, ensure_ascii=False)}")

                    # 通过MCP Client调用Server
                    result = await session.call_tool(func_name, func_args)
                    result_text = result.content[0].text

                    print(f"   结果: {result_text[:200]}...")

                    # 将结果加入上下文
                    messages.append({
                        "role": "assistant",
                        "content": content,
                    })
                    messages.append({
                        "role": "tool",
                        "content": result_text,
                    })

            else:
                print("⚠️ 达到最大迭代次数，Agent停止")


if __name__ == "__main__":
    query = "分析各部门的平均薪资水平，并找出销售额最高的3名员工及其所属部门"
    asyncio.run(run_agent(query))


2026年的前沿实践已经不满足于单个Agent，而是构建

。结合Google提出的A2A（Agent-to-Agent）协议，你可以让多个专精Agent各司其职：

用户请求: "生成本季度销售报告并发送邮件"
        │
        ▼
┌──────────────────┐
│  Coordinator Agent │  ← 任务分解与调度
└──────┬───────────┘
       │
  ┌────┴────┬────────────┐
  ▼         ▼            ▼
┌─────┐  ┌─────┐  ┌─────────┐
│DB   │  │Chart│  │ Email   │
│Agent│  │Agent│  │ Agent   │
│(MCP)│  │(MCP)│  │ (MCP)   │
└─────┘  └─────┘  └─────────┘
 查询数据   生成图表   发送邮件


每个Agent通过MCP连接各自的工具链，通过A2A协议互相通信——这就是2026年企业级AI应用的典型架构。

将MCP Server部署到生产环境时，务必关注以下要点：

从WAIC 2026的300+新品首发，到Claude Sonnet 5的深度MCP集成，2026年的AI行业正在经历一场深刻的范式转移：

AI不再只是"回答问题的大脑"，而是成为了"能够动手做事的完整智能体"

MCP协议就是这场变革的基础设施。它让每个开发者都能以极低成本，赋予AI操作真实世界的能力——读数据库、调API、写文件、发通知。正如Andrej Karpathy所说：

而现在，你手中的代码就是这张入场券。

【 2026年7月2日】 距离2026世界人工智能大会（WAIC 2026）开幕仅剩15天，上海世博中心正在紧锣密鼓地筹备这场被称为"AI界双十一"的年度盛会——超过300款AI新品将全球首发，涵盖大模型、具身智能、AI芯片等全产业链。与此同时，Anthropic于上周发布的Claude Sonnet 5再次引爆技术圈，其原生MCP（Model Context Protocol）深度集成能力，标志着AI智能体（Agent）生态正式从"概念验证"迈入"生产部署"阶段。

当AI学会"自己动手"：MCP协议如何让每个人都能构建智能体

【 2026年7月2日】 距离2026世界人工智能大会（WAIC 2026）开幕仅剩15天，上海世博中心正在紧锣密鼓地筹备这场被称为"AI界双十一"的年度盛会——超过300款AI新品将全球首发，涵盖大模型、具身智能、AI芯片等全产业链。与此同时，Anthropic于上周发布的Claude Sonnet 5再次引爆技术圈，其原生MCP（Model Context Protocol）深度集成能力，标志

人工智能

开发工具

后端

2026世界人工智能大会将首发300+AI新品，聚焦MCP协议推动AI智能体发展。MCP协议作为AI工具集成标准，实现一次适配多平台通用，支持数据库查询、文件操作等场景。文章详解如何构建MCPServer，让AI自主完成数据分析任务，并介绍多Agent协作架构等前沿实践。

数据分析

文件系统

PostgreSQL

智能体

数据库

ChatGPT

SQLite

路由

Python

Agent

MySQL

JSON

2026上云采购 | AI焕新·智启新局

cngw

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

当AI学会"自己动手"：MCP协议如何让每个人都能构建智能体

当AI学会"自己动手"：MCP协议如何让每个人都能构建智能体

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐