1.初始化项目 #

本项目基于 FastAPI 搭建，采用 SQLAlchemy 进行 ORM 数据库操作，适用于构建以「智能体」为中心的对话或多模型应用后端。以下为初始化项目与依赖配置的详细说明：

1. 创建虚拟环境与依赖安装
   推荐使用 uv 工具进行依赖和环境管理，兼容 npm/yarn 的开发体验，大幅加快 Python 依赖管理的效率。

   • 初始化项目目录（生成 pyproject.toml 等）:
     uv init
   • 安装所需依赖包：
     uv add python-multipart fastapi uvicorn[standard] sqlalchemy ...

2. 依赖说明

   • fastapi：主框架，支持异步 RESTful API 和自动生成 OpenAPI 文档，适合敏捷开发。
   • uvicorn[standard]：主流高性能的 ASGI 服务器，用于运行 FastAPI 项目，[standard]字样表示包含优化依赖如 uvloop、httptools 等，提高吞吐量和响应速度。
   • sqlalchemy：行业标准的 Python ORM，统一支持多种数据库，定义模型与数据结构。
   • pymysql：MySQL 适配驱动，配合 SQLAlchemy 实现数据存取。
   • cryptography：用于敏感信息的加解密保障安全。
   • pydantic/pydantic-settings：用于数据模型的类型验证和自动配置环境变量，与 FastAPI 深度集成。
   • httpx：现代化 HTTP 客户端，支持同步与异步请求，适配微服务或第三方 API 调用。
   • python-dotenv：方便本地开发时读取.env环境变量文件。
   • mcp[cli]：多模型协同平台的管理 CLI，可以集中便捷地管理各类 LLM 服务/插件/模型资源。
   • langchain & langchain-deepseek：打造基于大模型（如 DeepSeek、大语言模型等）的复杂推理、记忆和工具调用能力。

3. 目录结构建议

   ├── app/
   │   ├── __init__.py
   │   ├── main.py      # FastAPI 应用主入口
   │   ├── models.py    # ORM 数据模型定义
   │   └── database.py  # 数据库连接与Session管理
   ├── main.py          # 作为服务器启动入口（可选）
   ├── .env             # 环境变量（开发/部署配置）
   ├── requirements.txt / pyproject.toml
   └── README.md

4. 开发与启动

   • 初始化数据库表结构可通过 SQLAlchemy 自动生成。
   • 使用如下命令启动开发服务器（默认热重载）：

     uvicorn app.main:app --reload

   • 自定义配置（端口、host、日志等）可在 uvicorn.run() 或启动命令中指定。

uv init
uv add  python-multipart fastapi uvicorn[standard] sqlalchemy pymysql cryptography pydantic pydantic-settings httpx python-dotenv mcp[cli] langchain langchain-deepseek

2. 启动服务器 #

本项目推荐使用 FastAPI 作为 Web 服务主框架，并通过 Uvicorn 启动应用服务器，配合 SQLAlchemy 实现数据库管理。具体启动流程如下：

1. 本地开发环境准备
   确保 Python 环境和必需依赖已安装，可以参考前面提供的 uv add ... 安装命令。若有.env配置文件，建议同步环境变量。

2. 启动 API 服务方式

   • 方法一：直接使用 Uvicorn 命令行
     在项目根目录下运行：

     uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

     其中：

     • app.main:app 指定 FastAPI 实例的位置（app/main.py 里的 app 对象）。
     • --reload 让开发阶段的服务变动代码自动重启（建议开发环境开启，生产关闭）。
     • 可以通过 --port 设置监听端口。

   • 方法二：运行 main.py 脚本
     也可直接运行根目录下的 main.py，其内部会调用 uvicorn.run 启动服务（底层同样使用 Uvicorn）。

     python main.py

3. 验证服务启动情况
   启动后可访问 http://localhost:8000/docs 查看自动生成的 API 文档；
   访问 http://localhost:8000/health 可验证健康检查接口是否正常响应：

   {
     "status": "ok"
   }

4. 可选配置

   • 可通过 .env 文件自定义 FastAPI 的标题、描述、数据库地址等参数。
   • 若需绑定公网 IP 或自定义域名，请确保服务器开放相关端口。

注意：

• 生产部署建议关闭 --reload，并可结合多进程服务（如 Gunicorn 搭配 Uvicorn worker）。
• 若需对接前端/第三方系统，可提前规划跨域（CORS）、接口鉴权等中间件功能。

由此，可快速搭建起数据驱动、接口友好、可拓展的智能体后端服务。

2.1. init.py #

app/init.py

2.2. main.py #

app/main.py

# 导入FastAPI库
from fastapi import FastAPI

# 创建FastAPI应用实例，设置标题和版本号
app = FastAPI(title="智能体服务", version="0.1.0")

# 定义一个GET类型的/health路由用于健康检查
@app.get("/health")
def health():
    # 返回服务状态为ok的JSON响应
    return {"status": "ok"}

2.3. main.py #

main.py

# 导入uvicorn库，用于运行ASGI服务器
+import uvicorn

# 判断当前模块是否为主模块（直接运行该脚本时为True）
if __name__ == "__main__":
    # 启动uvicorn服务器，加载app.main模块中的app对象
    # host设置为0.0.0.0，允许外部访问
    # port设置为8000
    # reload=True表示代码变动时自动重启服务（开发环境常用）
+   uvicorn.run(
+       "app.main:app",
+       host="0.0.0.0",
+       port=8000,
+       reload=True,
+   )

3. 连接数据库 #

在「连接数据库」这一小节，我们将详细介绍如何在 FastAPI 项目中集成 SQLAlchemy 来操作 MySQL 数据库。

一般流程分为以下几步：

1. 配置数据库连接
   在 .env 和 app/config.py 中定义数据库连接字符串，便于统一管理和灵活切换。

   • .env 文件通过 DATABASE_URL 配置连接参数，不建议把账号密码硬编码到代码仓库。
   • 在 app/config.py 中通过 pydantic-settings 自动加载 .env 配置，提供便捷的全局访问。

2. 创建 SQLAlchemy 数据库引擎和会话

   • 在 app/database.py 中定义 engine（连接池）、SessionLocal（数据库会话工厂）和 Base（ORM模型基类）。
   • 所有 ORM 模型需继承自 Base，以便 SQLAlchemy 能自动发现和创建数据表。

3. 依赖注入数据库会话

   • 在业务路由中，需要为每个请求提供一个独立的数据库会话。通常可通过 FastAPI 的依赖注入机制（Depends）实现。
   • 推荐定义 get_db 函数：用 yield 生成 session，自动实现请求结束时的关闭和资源回收，避免连接泄漏。

4. 数据库表模型定义与迁移（可选）

   • 以 Python 类的方式定义数据表结构，继承自 Base。
   • 若需要自动迁移/同步表结构，可借助 Alembic 等工具完成。

5. 安全与性能建议

   • 生产环境下建议使用更复杂的数据库账号密码，并控制连接池数量。
   • 谨慎处理事务，避免长事务与死锁。

通过上述步骤，即可为智能体后端服务建立可靠的数据库访问能力，为后续业务开发打下坚实基础。

3.1 创建数据库 #

CREATE DATABASE ai_agent CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

3.2. .env #

.env

DATABASE_URL=mysql+pymysql://root:root@127.0.0.1:3306/ai_agent?charset=utf8mb4
CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173

3.3. config.py #

app/config.py

# 导入Path用于路径处理（虽然本文件未直接用到）
from pathlib import Path

# 导入pydantic_settings中的BaseSettings和SettingsConfigDict用于配置管理
from pydantic_settings import BaseSettings, SettingsConfigDict

# 定义Settings类，继承自BaseSettings，便于环境变量和配置文件的读取
class Settings(BaseSettings):
    # 配置SettingsConfigDict，指定.env文件路径及编码格式，并设置额外字段忽略
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

    # 数据库连接字符串，默认连接本地mysql数据库
    database_url: str = "mysql+pymysql://root:password@127.0.0.1:3306/agent?charset=utf8mb4"
    # 允许跨域的前端地址，多个地址以逗号分隔
    cors_origins: str = "http://localhost:5173,http://127.0.0.1:5173"

# 创建Settings实例，供项目其他部分导入使用
settings = Settings()

3.4. database.py #

app/database.py

# 导入SQLAlchemy的create_engine用于创建数据库引擎
from sqlalchemy import create_engine
# 导入sessionmaker用于会话创建，DeclarativeBase用于基类声明
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# 从app.config中导入settings对象，读取配置信息
from app.config import settings

# 定义ORM模型的基类，所有模型都将继承该类
class Base(DeclarativeBase):
    pass

# 创建数据库引擎，连接方式使用settings中的database_url
# pool_pre_ping=True用于防止数据库连接断开
# pool_recycle=3600设置连接池中连接的最大存活时间为3600秒
engine = create_engine(
    settings.database_url,
    pool_pre_ping=True,
    pool_recycle=3600,
)

# 创建会话工厂，autocommit=False表示手动提交事务
# autoflush=False表示不自动刷新
# bind=engine用于绑定数据库引擎
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 定义数据库会话的生成器，在依赖中使用
def get_db():
    # 创建一个数据库会话实例
    db = SessionLocal()
    try:
        # 使用yield返回会话对象
        yield db
    finally:
        # 关闭会话，释放资源
        db.close()

3.5. main.py #

main.py

# 导入uvicorn库，用于运行ASGI服务器
import uvicorn
# 从sqlalchemy导入text，用于执行原生SQL语句
+from sqlalchemy import text
# 从app.database模块导入get_db，用于获取数据库会话
+from app.database import get_db

# 尝试与数据库建立连接
+try:
    # 获取数据库会话生成器
+   db_gen = get_db()
    # 获取数据库会话实例
+   db = next(db_gen)
    # 执行一条简单的SQL语句以测试数据库连接
+   db.execute(text("SELECT 1"))
    # 打印数据库连接正常的信息
+   print("数据库连接正常")
# 如果发生异常
+except Exception as e:
    # 打印数据库连接失败的错误信息
+   print(f"数据库连接失败: {e}")
# 无论是否发生异常都会执行
+finally:
+   try:
        # 尝试关闭数据库会话生成器，释放资源
+       db_gen.close()
+   except Exception:
        # 如果关闭时发生异常则忽略
+       pass

# 判断是否为主程序运行入口
if __name__ == "__main__":
    # 启动uvicorn服务器，加载app.main模块下的app实例
    # host设置为0.0.0.0以便外部主机访问
    # port设置为8000
    # reload设置为True用于开发时自动重启
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
    )

4. 数据模型 #

以下是对数据模型的详细讲解： 本项目的数据模型主要使用SQLAlchemy的声明式ORM方式定义，所有模型都继承自 app.database 中的 Base 基类。

McpService 模型

用于存储第三方MCP服务的配置信息，主要字段说明如下：

• id：主键，自增，为每个服务分配唯一标识。
• name：服务名称，字符串类型，设置唯一约束（unique=True）和索引（index=True），保证每个服务名称唯一并提升查询效率。
• description：服务描述，可为 NULL，用于详细说明服务作用等扩展信息。
• protocol：协议类型，如 HTTP, MQTT 等，不能为空。
• config：用于保存服务的配置信息，采用 MySQL 的 JSON 类型字段，可灵活扩展不同服务的配置参数。

该表设计可拓展性强，新增服务类型或配置信息时无需改动表结构。

LlmModel 模型

用于存储大模型 API 提供方的信息和其支持的大模型列表，每个字段解释如下：

• id：主键，自增。
• provider_name：提供方名称，唯一且有索引，如 OpenAI、Azure。
• provider_icon：提供方的图标地址，可以为 NULL，便于前端展示。
• api_base_url：API 的基础请求地址，为必填项。
• api_key：用于访问该 LLM 提供方 API 的密钥，为必填项。
• api_key_url：密钥申请或获取地址，为可选项，便于新用户配置。
• model_names：存储提供方所支持的具体模型列表，数据类型为 JSON，例如 ["gpt-3.5-turbo", "gpt-4"]。

这种设计便于后续动态添加更多模型或多个提供方，同时支持前端读取支持模型以动态渲染 UI。

数据类型说明

• 采用 Mapped[...] 注解配合 mapped_column 明确字段类型，兼容 SQLAlchemy 2.0+。
• 字符串长度（如 String(255), String(1024)）根据实际业务场景预留充足长度。
• JSON 类型便于存储动态配置信息和模型名称列表（仅在支持 JSON 的数据库后端如 MySQL 5.7+ 有效）。
• 支持可空字段均加了 nullable=True 标记。

示例数据

• McpService 示例：

    {
      "id": 1,
      "name": "mqtt-service",
      "description": "用于IoT设备通讯的MQTT服务",
      "protocol": "MQTT",
      "config": {
        "host": "localhost",
        "port": 1883,
        "username": "user1",
        "password": "*****"
      }
    }

• LlmModel 示例：

    {
      "id": 1,
      "provider_name": "OpenAI",
      "provider_icon": "https://example.com/icon.png",
      "api_base_url": "https://api.openai.com/v1/",
      "api_key": "sk-****************",
      "api_key_url": "https://platform.openai.com/account/api-keys",
      "model_names": ["gpt-3.5-turbo", "gpt-4"]
    }

通过上述模型设计，可以方便扩展系统对接多种服务和大模型能力，并具备良好的灵活性与可维护性。

4.1. models.py #

app/models.py

# 导入datetime用于处理日期和时间
from datetime import datetime
# 导入SQLAlchemy中的字段类型和时间函数
from sqlalchemy import DateTime, String, Text, func
# 导入MySQL方言中的JSON类型
from sqlalchemy.dialects.mysql import JSON
# 导入ORM映射辅助工具
from sqlalchemy.orm import Mapped, mapped_column

# 导入数据库基类
from app.database import Base

# 定义McpService服务模型
class McpService(Base):
    # 设置对应的表名
    __tablename__ = "mcp_services"

    # 主键ID，自增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    # 服务名，唯一且有索引
    name: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    # 服务描述，允许为null
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    # 协议类型，必填
    protocol: Mapped[str] = mapped_column(String(32), nullable=False)
    # 配置信息，JSON格式，必填
    config: Mapped[dict] = mapped_column(JSON, nullable=False)

# 定义LlmModel大模型提供方模型
class LlmModel(Base):
    # 设置表名
    __tablename__ = "llm_models"

    # 主键ID，自增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    # 提供方名称，唯一且有索引
    provider_name: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    # 图标地址，允许为null
    provider_icon: Mapped[str | None] = mapped_column(Text, nullable=True)
    # API基础URL，必填
    api_base_url: Mapped[str] = mapped_column(String(1024), nullable=False)
    # API密钥，必填
    api_key: Mapped[str] = mapped_column(String(1024), nullable=False)
    # 密钥获取地址，可空
    api_key_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
    # 支持的模型名列表，JSON格式，必填
    model_names: Mapped[list[str]] = mapped_column(JSON, nullable=False)

# 定义Agent智能体模型
class Agent(Base):
    # 表名设置
    __tablename__ = "agents"

    # 主键、自增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    # 头像，可空
    avatar: Mapped[str | None] = mapped_column(Text, nullable=True)
    # 名称，有索引
    name: Mapped[str] = mapped_column(String(255), index=True)
    # 描述，可空
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    # 开场消息，可空
    opening_message: Mapped[str | None] = mapped_column(Text, nullable=True)
    # 系统提示，必填
    system_prompt: Mapped[str] = mapped_column(Text, nullable=False)
    # LLM提供方名称，必填
    llm_provider_name: Mapped[str] = mapped_column(String(255), nullable=False)
    # LLM模型名称，必填
    llm_model_name: Mapped[str] = mapped_column(String(255), nullable=False)
    # MCP服务ID列表，JSON格式，必填
    mcp_service_ids: Mapped[list[int]] = mapped_column(JSON, nullable=False)
    # 询问提示词（模板），可空
    ask_prompt_template: Mapped[str | None] = mapped_column(Text, nullable=True)
    # 询问变量，默认为空列表，JSON格式，不能为空
    ask_variables: Mapped[list[dict]] = mapped_column(JSON, nullable=False, default=list)

# 定义智能体对话会话模型
class AgentChatSession(Base):
    # 表名
    __tablename__ = "agent_chat_sessions"

    # 主键，自增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    # 智能体ID，有索引且必填
    agent_id: Mapped[int] = mapped_column(nullable=False, index=True)
    # 会话标题，必填，默认“新对话”
    title: Mapped[str] = mapped_column(String(255), nullable=False, default="新对话")
    # 创建时间，默认当前时间
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=False), server_default=func.now(), nullable=False
    )
    # 更新时间，默认当前时间，修改时更新
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=False), server_default=func.now(), onupdate=func.now(), nullable=False
    )

# 定义智能体对话消息模型
class AgentChatMessage(Base):
    # 表名
    __tablename__ = "agent_chat_messages"

    # 主键，自增
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    # 会话ID，有索引且必填
    session_id: Mapped[int] = mapped_column(nullable=False, index=True)
    # 发送者角色（如user/agent），必填
    role: Mapped[str] = mapped_column(String(32), nullable=False)
    # 消息内容，必填
    content: Mapped[str] = mapped_column(Text, nullable=False)
    # 附加元信息，默认为空dict，JSON格式
    meta: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    # 消息创建时间，默认为当前时间
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=False), server_default=func.now(), nullable=False
    )

4.2. 表说明 #

4.2.1 mcp_services #

4.2.2 llm_models #

4.2.3 agents #

4.2.4 agent_chat_sessions #

4.2.5 agent_chat_messages #

4.2.6 逻辑关系 #

4.3. ER 图 #

• agent_chat_sessions.agent_id → agents.id
• agent_chat_messages.session_id → agent_chat_sessions.id
• agents.llm_provider_name ↔ llm_models.provider_name（逻辑）
• agents.mcp_service_ids 内含 mcp_services.id（逻辑 N:M）

4.4. 高层关系 #

5. 建表 #

本节将详细说明如何通过 SQLAlchemy 数据模型管理上述数据表结构。

SQLAlchemy 数据模型定义

在 app/models.py 文件中，你需要为每个表定义一个对应的 SQLAlchemy ORM 类。例如：

from sqlalchemy import Column, Integer, String, Text, DateTime, JSON, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
import datetime

Base = declarative_base()

class Agent(Base):
    __tablename__ = "agents"

class AgentChatSession(Base):
    __tablename__ = "agent_chat_sessions"

class AgentChatMessage(Base):
    __tablename__ = "agent_chat_messages"

自动建表

• 自动建表：如 main.py 片段所示，Base.metadata.create_all(bind=engine) 可在应用启动时自动创建表结构。

关联关系说明

1. agent_chat_sessions.agent_id 通过外键指向 agents.id，实现一个 Agent 关联多个 Session 的 1:N 关系。
2. agent_chat_messages.session_id 通过外键指向 agent_chat_sessions.id，实现一个 Session 关联多条 Message 的 1:N 关系。
3. agents.llm_provider_name 与 llm_models.provider_name 通过名称字段逻辑绑定，无物理外键。
4. agents.mcp_service_ids 以 JSON 数组存储多个 mcp_services.id，实现灵活关联（N:M，需应用层逻辑管理）。

5.1. main.py #

app/main.py

# 导入FastAPI库
from fastapi import FastAPI
# 导入logging库以便后续日志记录
+import logging
# 导入asynccontextmanager用于异步上下文管理器
+from contextlib import asynccontextmanager
# 从app.database模块导入Base和engine，用于数据库相关操作
+from app.database import Base, engine
# 导入app.models模块下的所有内容（类、函数等）
+from app.models import * 
# 配置日志的基本设置，日志级别为INFO
+logging.basicConfig(level=logging.INFO)
# 定义一个异步上下文管理器，用于FastAPI生命周期
+@asynccontextmanager
+async def lifespan(app: FastAPI):
    # 创建所有数据库表结构（如果未存在则自动创建）
+   Base.metadata.create_all(bind=engine)
    # 通过yield挂起，等待应用关闭时进行清理
+   yield
# 创建FastAPI应用实例，设置API标题和版本号，并指定生命周期管理器
+app = FastAPI(title="智能体服务", version="0.1.0", lifespan=lifespan)
# 定义一个GET类型的/health路由用于健康检查
@app.get("/health")
def health():
    # 返回服务状态为ok的JSON响应
    return {"status": "ok"}

5.2. main.py #

main.py

# 导入uvicorn库，用于运行ASGI服务器
import uvicorn
# 判断是否为主程序运行入口
if __name__ == "__main__":
    # 启动uvicorn服务器，加载app.main模块下的app实例
    # host设置为0.0.0.0以便外部主机访问
    # port设置为8000
    # reload设置为True用于开发时自动重启
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
    )

6. 跨域 #

在FastAPI中实现跨域（CORS）支持，最常用的方法是引入CORSMiddleware中间件。这样可以确保你的API能够被浏览器中的前端应用安全地访问，尤其是在本地和线上环境存在不同域名或端口时。

• CORS（跨域资源共享）：默认情况下，浏览器出于安全考虑会阻止网页访问不同源（协议、域名或端口不同）下的API。CORS是一种机制，允许服务端声明可被哪些源访问，从而实现安全的跨域请求。
• CORSMiddleware：FastAPI中集成的中间件，配置后自动为API响应添加适当的CORS头部信息。

1. 导入中间件和配置
   • from fastapi.middleware.cors import CORSMiddleware
   • from app.config import settings
2. 解析 CORS 允许的来源
   • 通常会将允许的域名列表写在环境变量（示例：settings.cors_origins），用英文逗号分隔。
   • 通过列表推导式进行分割和清理空格、去除空字符串，得到最终的origins列表。
3. 注册中间件
   • 使用app.add_middleware(...)方法，把CORSMiddleware加到FastAPI应用上。
   • 通常建议设置：
     • allow_origins: 可访问的源组成的列表（如开发阶段通常允许所有源，生产环境请精确配置）。
     • allow_credentials: 是否允许cookie、认证等凭证。
     • allow_methods与allow_headers均设为["*"]，表示不限制方法和头部字段。

示例场景

• 前端（如本地 http://localhost:3000）开发时访问本后端API
• 生产环境下只允许公司域名访问API

此配置提升了服务的灵活性和安全性。

常见问题

• 配置了 CORS 但仍报跨域错误？请检查：
  • 前端请求地址（端口、协议等是否与允许列表对应）
  • allow_origins 是否包含了请求源
  • nginx、网关等外层代理是否覆盖或删改了CORS相关头信息

推荐做法

• 开发环境：cors_origins可设为*或http://localhost:3000等前端地址。
• 生产环境：cors_origins应精确枚举允许的正式域名，防止被恶意第三方利用。

6.1. main.py #

app/main.py

# 导入FastAPI库
from fastapi import FastAPI
# 导入logging库以便后续日志记录
import logging
# 导入asynccontextmanager用于异步上下文管理器
from contextlib import asynccontextmanager
# 导入FastAPI的CORS中间件，用于跨域资源共享
+from fastapi.middleware.cors import CORSMiddleware
# 导入项目配置settings对象
+from app.config import settings
# 从app.database模块导入Base和engine，用于数据库相关操作
from app.database import Base, engine
# 导入app.models模块下的所有内容（类、函数等）
from app.models import * 
# 配置日志的基本设置，日志级别为INFO
logging.basicConfig(level=logging.INFO)
# 定义一个异步上下文管理器，用于FastAPI生命周期
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 创建所有数据库表结构（如果未存在则自动创建）
    Base.metadata.create_all(bind=engine)
    # 通过yield挂起，等待应用关闭时进行清理
    yield
# 创建FastAPI应用实例，设置API标题和版本号，并指定生命周期管理器
app = FastAPI(title="智能体服务", version="0.1.0", lifespan=lifespan)
# 解析配置中的CORS来源列表，去除空白项和空字符串
+origins = [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
# 向FastAPI应用添加CORS中间件
+app.add_middleware(
+   CORSMiddleware,
    # 允许访问的来源列表，如果为空则允许所有来源（"*"）
+   allow_origins=origins or ["*"],
    # 允许携带cookie等凭证
+   allow_credentials=True,
    # 允许所有HTTP方法
+   allow_methods=["*"],
    # 允许所有HTTP头
+   allow_headers=["*"],
+)
# 定义一个GET类型的/health路由用于健康检查
@app.get("/health")
def health():
    # 返回服务状态为ok的JSON响应
    return {"status": "ok"}

7. 添加MCP服务 #

本节我们将为项目添加 MCP 服务 （即多通道处理服务，Multi-Channel Processing Service）。这一部分内容主要包括：MCP 服务的数据模型定义、接口（API）定义、数据库操作方法（Repository）实现、以及路由（Router）注册流程。

主要内容如下：

1. 数据模型（Model）和序列化（Schema）

   • 在 app/models.py 中增加 McpService 模型，描述 MCP 服务的数据结构及其字段（如名称、描述、协议类型、配置信息等）。
   • 在 app/schemas.py 中定义与模型对应的输入和输出序列化类，以用于数据校验和文档自动生成。

2. 数据库操作（Repository）

   • 在 app/repositories/mcp_repository.py 中实现 MCP 服务的增删查改方法。比如 create_mcp_service 函数，用于插入新的 MCP 服务记录，该函数会接收数据库会话对象和待插入的数据对象，处理后将数据持久化到数据库。

3. 接口路由（Router）

   • 路由文件（比如 app/routers/mcp.py）定义 MCP 服务的相关 API 接口，如创建服务、查询服务列表、获取详情、删除服务等。这些接口会调用 repository 层实现实际的数据操作。
   • 路由注册通常会在 FastAPI 主实例中（如 app/main.py）统一挂载。

4. 依赖注入与请求/响应模型

   • 结合 FastAPI 的依赖注入特性，通过参数注入 Session 数据库会话、以及参数校验自动对接相关 pydantic 数据模型。

5. 具体开发流程

   • 先设计并创建数据表和模型；
   • 再实现操作数据库的 repository 层方法；
   • 编写与之匹配的 schema；
   • 最后写 API 路由和接口逻辑，并将路由注册到应用主程序。

本节内容有助于你了解如何使用 FastAPI 构建结构清晰、解耦良好的 RESTful 服务，便于后续的功能扩展和维护。你可以根据需求灵活调整服务字段及接口设计。

7.1. init.py #

app/repositories/init.py

from . import mcp_repository

__all__ = ["mcp_repository"]

7.2. mcp_repository.py #

app/repositories/mcp_repository.py

# 导入SQLAlchemy的select（尽管本文件未使用，可以视情况保留或去除）
from sqlalchemy import select
# 导入SQLAlchemy的Session对象
from sqlalchemy.orm import Session

# 导入项目中的models模块
from app import models
# 导入项目中的schemas模块
from app import schemas

# 定义创建McpService的函数，接收数据库会话db和待创建数据data，返回新建的McpService对象
def create_mcp_service(db: Session, data: schemas.McpServiceCreate) -> models.McpService:
    # 构造McpService模型对象，strip去除名字首尾空白
    row = models.McpService(
        name=data.name.strip(),#去除名字首尾空白
        description=data.description,#服务描述
        protocol=data.protocol.value,#协议类型
        config=data.config,#配置信息
    )
    # 添加新对象到会话
    db.add(row)
    # 提交事务，将更改保存到数据库
    db.commit()
    # 刷新实例，确保row包含数据库自动生成的字段值
    db.refresh(row)
    # 返回新建的McpService对象
    return row

7.3. init.py #

app/routers/init.py

# routers

7.4. mcp_services.py #

app/routers/mcp_services.py

# 引入日志模块
import logging

# 从FastAPI导入路由器、依赖项和HTTP异常类
from fastapi import APIRouter, Depends, HTTPException
# 从SQLAlchemy导入唯一性错误异常
from sqlalchemy.exc import IntegrityError
# 从SQLAlchemy导入ORM会话对象
from sqlalchemy.orm import Session

# 导入应用程序的数据模型
from app import schemas
# 导入mcp_repository模块，包含数据库操作方法
from app.repositories import mcp_repository
# 导入用于获取数据库会话的依赖函数
from app.database import get_db

# 创建API路由器，设置前缀和标签
router = APIRouter(prefix="/api/mcp-services", tags=["mcp-services"])
# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)

# 定义创建服务的POST接口，响应模型为McpServiceOut
@router.post("", response_model=schemas.McpServiceOut)
def create_service(payload: schemas.McpServiceCreate, db: Session = Depends(get_db)):
    # 使用try-except来捕捉数据库插入冲突
    try:
        # 调用仓库方法创建服务
        return mcp_repository.create_mcp_service(db, payload)
    # 捕获唯一性约束异常（如名称重复）
    except IntegrityError:
        # 回滚数据库会话，撤消操作
        db.rollback()
        # 抛出409冲突异常，返回错误信息
        raise HTTPException(status_code=409, detail="名称已存在")

7.5. schemas.py #

app/schemas.py

# 导入枚举类型
from enum import Enum
# 导入Any类型用于类型注解
from typing import Any

# 从pydantic导入BaseModel、Field和field_validator用于数据验证
from pydantic import BaseModel, Field, field_validator

# 定义MCP协议类型的枚举类
class McpProtocol(str, Enum):
    # MCP 协议类型注释
    """MCP 协议类型"""
    # stdio协议
    stdio = "stdio"
    # streamable-http协议
    streamable_http = "streamable-http"
    # sse协议
    sse = "sse"

# 定义MCP服务基础模型
class McpServiceBase(BaseModel):
    # MCP 服务基础模型注释
    """MCP 服务基础模型"""
    # 服务名称，字符串类型，长度1-255
    name: str = Field(..., min_length=1, max_length=255)
    # 服务描述，可选字段，字符串或None
    description: str | None = None
    # 协议类型，使用McpProtocol枚举
    protocol: McpProtocol
    # 配置信息，要求为字典类型
    config: dict[str, Any]

    # 对config字段添加验证器，确保其为字典类型
    @field_validator("config", mode="before")
    @classmethod
    def config_not_empty(cls, v: Any) -> Any:
        # 验证 config 是否为 JSON 对象，如果不是字典则抛出异常
        """验证 config 是否为 JSON 对象"""
        if not isinstance(v, dict):
            raise ValueError("config 必须为 JSON 对象")
        return v

# 定义MCP服务创建模型，继承自McpServiceBase
class McpServiceCreate(McpServiceBase):
    # 不增加额外内容，直接继承
    pass

# 定义MCP服务输出模型
class McpServiceOut(BaseModel):
    # ID字段，整型
    id: int
    # 服务名称
    name: str
    # 服务描述，可选字段
    description: str | None
    # 协议类型，字符串
    protocol: str
    # 配置信息，字典类型
    config: dict[str, Any]

    # 设置模型配置，允许从ORM对象属性读取数据
    model_config = {"from_attributes": True}

7.6. main.py #

app/main.py

# 导入FastAPI库
from fastapi import FastAPI
# 导入logging库以便后续日志记录
import logging
# 导入asynccontextmanager用于异步上下文管理器
from contextlib import asynccontextmanager
# 导入FastAPI的CORS中间件，用于跨域资源共享
from fastapi.middleware.cors import CORSMiddleware
# 导入项目配置settings对象
from app.config import settings
# 从app.database模块导入Base和engine，用于数据库相关操作
from app.database import Base, engine
# 导入app.models模块下的所有内容（类、函数等）
from app.models import * 
# 导入MCP服务路由
+from app.routers import mcp_services
# 配置日志的基本设置，日志级别为INFO
logging.basicConfig(level=logging.INFO)
# 定义一个异步上下文管理器，用于FastAPI生命周期
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 创建所有数据库表结构（如果未存在则自动创建）
    Base.metadata.create_all(bind=engine)
    # 通过yield挂起，等待应用关闭时进行清理
    yield
# 创建FastAPI应用实例，设置API标题和版本号，并指定生命周期管理器
app = FastAPI(title="智能体服务", version="0.1.0", lifespan=lifespan)
# 解析配置中的CORS来源列表，去除空白项和空字符串
origins = [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
# 向FastAPI应用添加CORS中间件
app.add_middleware(
    CORSMiddleware,
    # 允许访问的来源列表，如果为空则允许所有来源（"*"）
    allow_origins=origins or ["*"],
    # 允许携带cookie等凭证
    allow_credentials=True,
    # 允许所有HTTP方法
    allow_methods=["*"],
    # 允许所有HTTP头
    allow_headers=["*"],
)
# 包含MCP服务路由
+app.include_router(mcp_services.router)
# 定义一个GET类型的/health路由用于健康检查
@app.get("/health")
def health():
    # 返回服务状态为ok的JSON响应
    return {"status": "ok"}