异步REST API工具

FastAPI + MCP 是一个现代的异步REST API,具有MCP风格的工具调用层,适用于需要同时支持人类用户和AI代理的服务场景。

  • by Sheikhharis50
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Sheikhharis50/fastapi-mcp

使用说明

FastAPI + MCP

A modern async REST API with an MCP-style tool invocation layer — built with FastAPI, SQLAlchemy 2, and Pydantic v2.


概述

fastapi mcp 在同一服务层上公开两个接口:

接口 说明
REST API (/users) 用于用户管理的传统CRUD端点
MCP工具调度程序 (/mcp/invoke) 一个单一的端点,用经过验证的输入路由命名的工具调用——灵感来自模型上下文协议模式

这两个接口共享相同的异步服务层、数据库模型和Pydantic模式,展示了单个后端如何为人类和人工智能代理提供服务。

特性

  • 完全异步 --异步SQLAlchemy引擎+用于非阻塞I/O的aiosqlite
  • 双接口 --REST端点 和 从一个代码库调用MCP风格的工具
  • Pydantic v2验证 --严格的模式 EmailStr长度约束,以及 ConfigDict
  • 统一响应格式 --每个REST响应都包装在一致的 ApiResponse 信封
  • 请求时间 --中间件注入 X-Process-Time 每个响应的标题
  • 自动配置数据库 --通过生命周期挂钩在启动时创建SQLite数据库
  • 交互式文档 --Swagger用户界面位于 /docs ReDoc /redoc 开箱即用

技术栈

层 技术
框架 快速API + Uvicorn
ORM SQL炼金术2 (异步)
数据库 SQLite通过 aiosqlite
验证 Pydantic v2
包管理器 紫外线
Linting 拉夫 + 排序

项目结构

app/
├── main.py              # FastAPI app, lifespan, middleware & router registration
├── db.py                # Async engine, session factory, Base
├── middlewares/
│   └── timer.py         # X-Process-Time header middleware
├── models/
│   └── user.py          # SQLAlchemy User model
├── routers/
│   ├── api.py           # REST /users endpoints
│   └── mcp.py           # POST /mcp/invoke tool dispatcher
├── schemas/
│   ├── api.py           # ApiResponse envelope
│   └── user.py          # UserCreate, UserUpdate, UserResponse
└── services/
    └── user.py          # Async CRUD business logic

入门指南

先决条件

安装

# Clone the repo
git clone https://github.com/Sheikhharis50/fastapi-mcp.git
cd fastapi-mcp

# Install dependencies with uv
uv sync

# — or with pip —
pip install -e .

运行服务器

uvicorn app.main:app --reload

API现已在 http://127.0.0.1:8000.

URL 描述
http://127.0.0.1:8000/docs Swagger用户界面
http://127.0.0.1:8000/redoc 重新记录
http://127.0.0.1:8000/health 健康检查

api参考

REST端点

方法 端点 描述
GET /health 健康检查
POST /users/ 创建新用户
GET /users/ 列出所有用户
GET /users/{user_id} 按ID获取用户
PUT /users/{user_id} 更新用户
DELETE /users/{user_id} 删除用户

示例——创建用户

curl -X POST http://127.0.0.1:8000/users/ \
  -H "Content-Type: application/json" \
  -d '{"name": "Haris", "email": "haris@example.com"}'
{
  "message": "User created successfully",
  "data": {
    "id": 1,
    "name": "Haris",
    "email": "haris@example.com"
  }
}

MCP工具调用

发送a POST 请求 /mcp/invoke 带有工具名称和输入有效载荷:

curl -X POST http://127.0.0.1:8000/mcp/invoke \
  -H "Content-Type: application/json" \
  -d '{"tool": "create_user", "input": {"name": "Haris", "email": "haris@example.com"}}'

可用工具

工具 输入 描述
create_user { "name": str, "email": str } 创建新用户
get_user { "user_id": int } 按ID获取用户
list_users {} 列出所有用户
update_user { "user_id": int, "name": str, "email": str } 更新现有用户
delete_user { "user_id": int } 删除用户

返回无效的工具名称 400,并返回模式冲突 422 带有Pydantic错误详细信息。

建筑

Client (REST)                Client (MCP / AI Agent)
      │                                │
      ▼                                ▼
  /users/*                      /mcp/invoke
      │                                │
      ▼                                ▼
  api.py router              mcp.py router
      │                          │
      │    ┌─────────────────────┘
      ▼    ▼
  services/user.py    ← shared business logic
        │
        ▼
  models/user.py + db.py   ← async SQLAlchemy + SQLite

贡献

  1. 分叉存储库
  2. 创建功能分支(git checkout -b feature/amazing-feature)
  3. 提交您的更改(git commit -m 'Add amazing feature')
  4. 推到分支(git push origin feature/amazing-feature)
  5. 打开拉取请求

作者

谢赫·哈里斯·扎希德 — @谢赫哈尔50