Skip to main content

接口开发

API概述​

API，全称为 应用程序编程接口（Application Programming Interface），是一组定义了不同软件组件之间如何交互的规则和协议。API 允许不同的软件系统通过预定义的接口进行通信，使得开发者能够利用现有的功能和服务，而无需从头开始构建。这种模块化的设计不仅提高了开发效率，还促进了软件生态系统的扩展和创新。

一个典型的 API 包含以下几个关键组成部分：

  • 请求（Request）： 客户端向服务器发送的调用请求，通常包含请求的方法、URL、头信息和参数。
  • 响应（Response）： 服务器对客户端请求的返回结果，包含状态码、头信息和数据载体。
  • 端点（Endpoint）： API 提供的具体功能接口的 URL 地址。
  • 方法（Method）： 定义在特定端点上可以执行的操作，如 GET、POST、PUT、DELETE 等。
  • 数据格式（Data Format）： 传输数据的格式，常见的有 JSON、XML 等。

接口标准​

特性OpenAI API 风格MCPOpenAPI
类型商业 API 设计开放通信协议开放 API 描述规范
是否开放治理❌（OpenAI 控制）✅（社区驱动）✅（Linux 基金会）
是否可自由实现⚠️（法律风险模糊）✅（MIT 许可）✅
是否有正式规范文档✅（但非标准组织发布）✅（mcp.spec）✅（openapis.org）
目标调用 LLMLLM 与工具通信描述 RESTful API

开发框架​

任何一家公司都可以定义符合自己公司业务需求的 API，但是他们可能使用的都是同一套开发框架在 Python 中，有多种接口开发框架可供选择：

维度FastAPIDjango
发布时间2018 年2005 年
开发语言PythonPython
架构风格微框架，基于 Starlette 和 Pydantic大型框架，支持全栈开发，内置 ORM、Admin 等
性能高，基于 ASGI，支持异步特性中等，依赖于 WSGI
异步支持原生支持，使用 async/await通过 Django 3.0+ 开始支持异步（仍在逐步完善）
数据验证内置 Pydantic，支持强类型和自动验证依赖于表单类（如 Forms 和 DRF 的 serializers）
自动文档生成内置支持 OpenAPI 和 Swagger UI不原生支持，需依赖 Django REST Framework (DRF)
学习曲线中等，需了解类型注解和异步编程较高，包含多种内置组件，适合系统性学习
社区和生态新兴框架，社区成长迅速，但生态略小最悠久且成熟，生态完善，适合大型项目
可扩展性高，强大的原生功能减少了扩展需求中等，内置组件强大，但灵活性相对较低
依赖注入支持原生支持依赖注入无原生支持
生产就绪稳定性较高，但相比 Flask 和 Django 使用年限较短成熟稳定，被广泛应用于大型生产系统
调试工具支持 Starlette 的调试功能内置调试模式，适合开发测试
使用场景高性能、异步处理或自动文档的场景大型系统开发，支持复杂业务逻辑和多用户系统
ORM 支持无原生支持，需借助 Tortoise-ORM 等原生支持，提供 Django ORM
Admin 界面无，需要自行开发原生支持，提供强大的 Admin 管理后台
支持的 Python 版本Python 3.6+Python 3.8+
对类型注解的支持原生支持，强类型友好不直接支持，但可结合第三方库使用

FastAPI​

安装：pip install "fastapi[standard]"

基础用法​

相较于 Flask，FastAPI 会自动生成 2 种风格的 API 文档，地址为：http://127.0.0.1:8000/docs 和 http://127.0.0.1:8000/redoc

from fastapi import FastAPI, Header, HTTPException, Query, Path
from pydantic import BaseModel
from typing import Optional, Dict

app = FastAPI()

# Pydantic Models
class PostExampleModel(BaseModel):
    title: Optional[str] = "默认标题"
    content: Optional[str] = "默认内容"

class PutExampleModel(BaseModel):
    name: Optional[str] = "默认名称"
    description: Optional[str] = "默认描述"

class ApiResourcePostModel(BaseModel):
    key: str

class ApiResourcePutModel(BaseModel):
    field: str

# GET /get_example/{user_id}
@app.get("/get_example/{user_id}")
def get_example(
    user_id: int,
    name: Optional[str] = Query("默认名字"),
    user_agent: Optional[str] = Header("未知")
):
    """
    **方法：** GET

    **示例请求：**
    http:day3.ipynb
    GET /get_example/123?name=Alice HTTP/1.1
    Host: example.com
    User-Agent: Mozilla/5.0

    **示例响应：**
    json
    {
        "user_id": 123,
        "name": "Alice",
        "user_agent": "Mozilla/5.0"
    }

    """
    return {"user_id": user_id, "name": name, "user_agent": user_agent}

# POST /post_example
@app.post("/post_example")
def post_example(
    data: PostExampleModel,
    authorization: Optional[str] = Header("无授权")
):
    """
    **方法：** POST

    **示例请求：**
    http
    POST /post_example HTTP/1.1
    Host: example.com
    Content-Type: application/json
    Authorization: Bearer token

    {
        "title": "示例标题",
        "content": "示例内容"
    }

    **示例响应：**
    json
    {
        "title": "示例标题",
        "content": "示例内容",
        "authorization": "Bearer token"
    }

    """
    return {"title": data.title, "content": data.content, "authorization": authorization}

# PUT /put_example/{item_id}
@app.put("/put_example/{item_id}")
def put_example(
    item_id: int,
    data: PutExampleModel,
    x_custom_header: Optional[str] = Header("无自定义头")
):
    """
    **方法：** PUT

    **示例请求：**
    http
    PUT /put_example/456 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded
    X-Custom-Header: 自定义值

    name=新名称&description=新描述

    **示例响应：**
    json
    {
        "item_id": 456,
        "name": "新名称",
        "description": "新描述",
        "custom_header": "自定义值"
    }

    """
    return {
        "item_id": item_id,
        "name": data.name,
        "description": data.description,
        "custom_header": x_custom_header,
    }

# DELETE /delete_example
@app.delete("/delete_example")
def delete_example(
    confirm: bool = Query(False),
    authorization: Optional[str] = Header("无授权")
):
    """
    **方法：** DELETE

    **示例请求（确认删除）：**
    http
    DELETE /delete_example?confirm=true HTTP/1.1
    Host: example.com
    Authorization: Bearer token

    **示例响应（确认删除）：**
    json
    {
        "deleted": true,
        "authorization": "Bearer token"
    }


    **示例请求（未确认删除）：**
    http
    DELETE /delete_example HTTP/1.1
    Host: example.com

    **示例响应（未确认删除）：**
    json
    {
        "message": "删除未确认"
    }

    HTTP 状态码：400
    """
    if not confirm:
        raise HTTPException(status_code=400, detail="删除未确认")
    return {"deleted": confirm, "authorization": authorization}

# GET /api/resource/{resource_id}
@app.get("/api/resource/{resource_id}")
def api_resource_get(
    resource_id: str,
    detail: Optional[str] = Query("basic")
):
    """
    **方法：** GET

    **请求示例：**
    GET /api/resource/abc123?detail=full HTTP/1.1
    Host: example.com

    **响应示例：**
    json
    {
        "method": "GET",
        "resource_id": "abc123",
        "detail": "full"
    }

    """
    return {"method": "GET", "resource_id": resource_id, "detail": detail}

# POST /api/resource/{resource_id}
@app.post("/api/resource/{resource_id}")
def api_resource_post(
    resource_id: str,
    data: ApiResourcePostModel
):
    """
    **方法：** POST

    **请求示例：**
    POST /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/json

    {
        "key": "value"
    }

    **响应示例：**
    json
    {
        "method": "POST",
        "resource_id": "abc123",
        "data": {
            "key": "value"
        }
    }

    """
    return {"method": "POST", "resource_id": resource_id, "data": data.dict()}

# PUT /api/resource/{resource_id}
@app.put("/api/resource/{resource_id}")
def api_resource_put(
    resource_id: str,
    data: ApiResourcePutModel
):
    """
    **方法：** PUT

    **请求示例：**
    PUT /api/resource/abc123 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded

    field=new%20value

    **响应示例：**
    json
    {
        "method": "PUT",
        "resource_id": "abc123",
        "data": {
            "field": "new value"
        }
    }

    """
    return {"method": "PUT", "resource_id": resource_id, "data": data.dict()}

# DELETE /api/resource/{resource_id}
@app.delete("/api/resource/{resource_id}")
def api_resource_delete(
    resource_id: str,
    confirm: bool = Query(False)
):
    """
    **方法：** DELETE

    **请求示例：**
    DELETE /api/resource/abc123?confirm=true HTTP/1.1
    Host: example.com

    **响应示例：**
    json
    {
        "method": "DELETE",
        "resource_id": "abc123",
        "confirm": true
    }

    """
    return {"method": "DELETE", "resource_id": resource_id, "confirm": confirm}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

下面是异步函数作为 FastAPI 的返回值的示例

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import random

app = FastAPI()

async def async_range(n):
    for i in range(n):
        yield f"<p>{i}</p>"
        await asyncio.sleep(random.randint(1, 3) * 1)

@app.get("/stream")
async def stream_numbers():
    return StreamingResponse(async_range(5), media_type="html", background=None)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

传输二进制​

接口除了可以传输文本数据，还可传输图片、视频等数据，OpenCV 可以捕获屏幕，将两者结合起来，实现内网直播功能，可以在局域网内通过浏览器观看屏幕共享。


import os
from importlib import import_module
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
import uvicorn

from io import BytesIO
import cv2
from PIL import ImageGrab, Image
import time
import threading
from threading import get_ident


class CameraEvent(object):
    def __init__(self):
        self.events = {}

    def wait(self):
        ident = get_ident()
        if ident not in self.events:
            self.events[ident] = [threading.Event(), time.time()]
        return self.events[ident][0].wait()

    def set(self):
        now = time.time()
        remove = None
        for ident, event in self.events.items():
            if not event[0].isSet():
                event[0].set()
                event[1] = now
            else:
                if now - event[1] > 5:
                    remove = ident
        if remove:
            del self.events[remove]

    def clear(self):
        self.events[get_ident()][0].clear()


class BaseCamera(object):
    thread = None
    frame = None
    last_access = 0
    event = CameraEvent()

    def __init__(self):
        if BaseCamera.thread is None:
            BaseCamera.last_access = time.time()

            BaseCamera.thread = threading.Thread(target=self._thread)
            BaseCamera.thread.start()

            while self.get_frame() is None:
                time.sleep(0)

    def get_frame(self):
        BaseCamera.last_access = time.time()

        BaseCamera.event.wait()
        BaseCamera.event.clear()

        return BaseCamera.frame

    @staticmethod
    def frames():
        raise RuntimeError('Must be implemented by subclasses.')

    @classmethod
    def _thread(cls):
        print('Starting camera thread.')
        frames_iterator = cls.frames()
        for frame in frames_iterator:
            BaseCamera.frame = frame
            BaseCamera.event.set()
            time.sleep(0)
            if time.time() - BaseCamera.last_access > 10:
                frames_iterator.close()
                print('Stopping camera thread due to inactivity.')
                break
        BaseCamera.thread = None



class Camera(BaseCamera):
    video_source = 0

    @staticmethod
    def set_video_source(source):
        Camera.video_source = source

    @staticmethod
    def frames():
        camera = cv2.VideoCapture(Camera.video_source)
        if not camera.isOpened():
            raise RuntimeError('Error')

        while True:
            image = ImageGrab.grab()  # 获取屏幕数据
            # w, h = image.size
            image = image.resize((1366, 750), Image.LANCZOS)  # 图片缩放
            output_buffer = BytesIO()  # 创建二进制对象
            image.save(output_buffer, format='JPEG', quality=100)  # quality提升图片分辨率
            frame = output_buffer.getvalue()  # 获取二进制数据
            yield frame  # 生成器返回一张图片的二进制数据

app = FastAPI()


@app.get('/', response_class=HTMLResponse)
async def index():
    """
    视图函数
    :return:
    """
    return '''<html>

<head>
    <title>屏幕共享</title>
</head>

<body>
    <img src="/video_feed">
</body>

</html>'''


def gen(camera):
    """
    流媒体发生器
    """
    while True:
        frame = camera.get_frame()

        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')


@app.get('/video_feed')
async def video_feed():
    """流媒体数据"""
    return StreamingResponse(gen(Camera()),
                            media_type='multipart/x-mixed-replace; boundary=frame')


if __name__ == '__main__':
    ip_host = '127.0.0.1'  # 本机ip地址
    ip_host2 = '0.0.0.0'  # 内网ip地址
    import webbrowser
    webbrowser.open(f'http://{ip_host}:80')
    uvicorn.run(app, host=ip_host2, port=80)

MCP​

Model Context Protocol（MCP） 是一种开放协议，旨在标准化应用程序与大型语言模型（LLM）之间的上下文交互。

MCP最初设计的对象是各类AI客户端（如 Claude Desktop、Cursor）,客户端通过配置文件连接 MCP 服务器。

MCP 服务器可以暴露多种类型的功能：

  • Resources（资源）： 只读数据源，如文件内容、数据库查询结果等
  • Tools（工具）： LLM 可以调用的函数，用于执行操作或获取动态数据
  • Prompts（提示词）： 预定义的提示词模板，可供 LLM 使用
  • Sampling(采样)： 工具可反向调用大模型为自己生成一些内容

并不是所有的主要类型功能都被支持，其中Tools支持范围最广。

为了更快速的构建MCP，可以使用FastMCP 处理所有复杂的协议细节，专注于构建。大多数情况下，只需装饰一个 Python 函数即可，FastMCP 会处理剩下的工作。

参考地址：https://github.com/jlowin/fastmcp

代码示例​

file_server.py
"""
FastMCP Echo Server
"""

from fastmcp import FastMCP

# Create server
mcp = FastMCP("Echo Server")


@mcp.tool
def echo_tool(text: str) -> str:
    """Echo the input text"""
    return text


@mcp.resource("echo://static")
def echo_resource() -> str:
    return "Echo!"


@mcp.resource("echo://{text}")
def echo_template(text: str) -> str:
    """Echo the input text"""
    return f"Echo: {text}"


@mcp.prompt("echo")
def echo_prompt(text: str) -> str:
    return text

配置文件​

上面的代码我部署在https://fastmcp.cloud/中，下面的地址是其为我随机分配的。填入cursor后登录验证即可。

{
  "mcpServers": {
    "fastmcp": {
      "url": "https://magnificent-crimson-antlion.fastmcp.app/mcp",
      "transport": "http"
    }
  }
}

如果是通过放在本地，则只需要配置全局的fastmcp,确保路径正确即可。

{
  "mcpServers": {
    "local-echo-server": {
      "command": "fastmcp",
      "args": ["run", "C:\\Users\\allen\\Desktop\\MCP\\server.py"],
      "transport": "stdio"
    }
  }
}

更多信息请参考 MCP 官方文档 和 PyPI 包页面。

Django​

基础用法​

Django 的接口开发，需要先创建项目和应用，然后创建视图函数，最后在 urls.py 中配置路由。

虽然 Django 的接口开发比较繁琐，但是 Django 的接口开发比较规范，易于维护。

安装 Django 和 Django REST framework

pip install django djangorestframework

创建项目和应用

django-admin startproject myproject
cd myproject
django-admin startapp apiapp

项目结构如下：

myproject/
├── apiapp/
│   ├── __init__.py
│   ├── views.py
│   ├── urls.py
│   └── serializers.py
├── myproject/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── manage.py

在 myproject/settings.py 中添加 rest_framework 和 apiapp 到 INSTALLED_APPS 列表中：

INSTALLED_APPS = [
    ... # 保持原有配置
    'rest_framework',
    'apiapp',
]

定义序列化器，有助于验证和解析输入数据。

apiapp/serializers.py
from rest_framework import serializers

class PostExampleSerializer(serializers.Serializer):
    title = serializers.CharField(default="默认标题", required=False)
    content = serializers.CharField(default="默认内容", required=False)

class PutExampleSerializer(serializers.Serializer):
    name = serializers.CharField(default="默认名称", required=False)
    description = serializers.CharField(default="默认描述", required=False)

class ApiResourcePostSerializer(serializers.Serializer):
    key = serializers.CharField()

class ApiResourcePutSerializer(serializers.Serializer):
    field = serializers.CharField()

定义视图函数

apiapp/views.py
from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework.views import APIView

from .serializers import (
    PostExampleSerializer,
    PutExampleSerializer,
    ApiResourcePostSerializer,
    ApiResourcePutSerializer,
)

@api_view(['GET'])
def get_example(request, user_id):
    """
    **方法：** GET

    **示例请求：**
    GET /get_example/123?name=Alice HTTP/1.1
    Host: example.com
    User-Agent: Mozilla/5.0

    **示例响应：**    ```json
    {
        "user_id": 123,
        "name": "Alice",
        "user_agent": "Mozilla/5.0"
    }
    """
    name = request.query_params.get('name', '默认名字')
    user_agent = request.headers.get('User-Agent', '未知')
    return Response({"user_id": user_id, "name": name, "user_agent": user_agent})

@api_view(['POST'])
def post_example(request):
    """
    **方法：** POST

    **示例请求：**
    POST /post_example HTTP/1.1
    Host: example.com
    Content-Type: application/json
    Authorization: Bearer token

    {
        "title": "示例标题",
        "content": "示例内容"
    }

    **示例响应：**    ```json
    {
        "title": "示例标题",
        "content": "示例内容",
        "authorization": "Bearer token"
    }
    """
    serializer = PostExampleSerializer(data=request.data)
    if serializer.is_valid():
        title = serializer.validated_data.get('title', '默认标题')
        content = serializer.validated_data.get('content', '默认内容')
        auth = request.headers.get('Authorization', '无授权')
        return Response({"title": title, "content": content, "authorization": auth})
    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['PUT'])
def put_example(request, item_id):
    """
    **方法：** PUT

    **示例请求：**
    PUT /put_example/456 HTTP/1.1
    Host: example.com
    Content-Type: application/x-www-form-urlencoded
    X-Custom-Header: 自定义值

    name=新名称&description=新描述

    **示例响应：**    ```json
    {
        "item_id": 456,
        "name": "新名称",
        "description": "新描述",
        "custom_header": "自定义值"
    }
    """
    serializer = PutExampleSerializer(data=request.data)
    if serializer.is_valid():
        name = serializer.validated_data.get('name', '默认名称')
        description = serializer.validated_data.get('description', '默认描述')
        custom_header = request.headers.get('X-Custom-Header', '无自定义头')
        return Response({
            "item_id": item_id,
            "name": name,
            "description": description,
            "custom_header": custom_header,
        })
    return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['DELETE'])
def delete_example(request):
    """
    **方法：** DELETE

    **示例请求（确认删除）：**
    DELETE /delete_example?confirm=true HTTP/1.1
    Host: example.com
    Authorization: Bearer token

    **示例响应（确认删除）：**    ```json
    {
        "deleted": true,
        "authorization": "Bearer token"
    }    ```

    **示例请求（未确认删除）：**
    DELETE /delete_example HTTP/1.1
    Host: example.com

    **示例响应（未确认删除）：**    ```json
    {
        "message": "删除未确认"
    }    ```
    HTTP 状态码：400
    """
    confirm = request.query_params.get('confirm', 'false').lower() == 'true'
    auth = request.headers.get('Authorization', '无授权')
    if not confirm:
        return Response({"message": "删除未确认"}, status=status.HTTP_400_BAD_REQUEST)
    return Response({"deleted": confirm, "authorization": auth})

class ApiResourceView(APIView):
    """
    **方法：** GET, POST, PUT, DELETE

    **GET 请求示例：**
    GET /api/resource/abc123?detail=full HTTP/1.1
    Host: example.com

    **GET 响应示例：**    ```json
    {
        "method": "GET",
        "resource_id": "abc123",
        "detail": "full"
    }
    """

    def get(self, request, resource_id):
        detail = request.query_params.get('detail', 'basic')
        return Response({"method": "GET", "resource_id": resource_id, "detail": detail})

    def post(self, request, resource_id):
        """
        **POST 请求示例：**
        POST /api/resource/abc123 HTTP/1.1
        Host: example.com
        Content-Type: application/json

        {
            "key": "value"
        }

        **POST 响应示例：**        ```json
        {
            "method": "POST",
            "resource_id": "abc123",
            "data": {
                "key": "value"
            }
        }        ```
        """
        serializer = ApiResourcePostSerializer(data=request.data)
        if serializer.is_valid():
            data = serializer.validated_data
            return Response({"method": "POST", "resource_id": resource_id, "data": data})
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def put(self, request, resource_id):
        """
        **PUT 请求示例：**
        PUT /api/resource/abc123 HTTP/1.1
        Host: example.com
        Content-Type: application/x-www-form-urlencoded

        field=new%20value

        **PUT 响应示例：**        ```json
        {
            "method": "PUT",
            "resource_id": "abc123",
            "data": {
                "field": "new value"
            }
        }        ```
        """
        serializer = ApiResourcePutSerializer(data=request.data)
        if serializer.is_valid():
            data = serializer.validated_data
            return Response({"method": "PUT", "resource_id": resource_id, "data": data})
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def delete(self, request, resource_id):
        """
        **DELETE 请求示例：**
        DELETE /api/resource/abc123?confirm=true HTTP/1.1
        Host: example.com

        **DELETE 响应示例：**        ```json
        {
            "method": "DELETE",
            "resource_id": "abc123",
            "confirm": true
        }        ```
        """
        confirm = request.query_params.get('confirm', 'false').lower() == 'true'
        return Response({"method": "DELETE", "resource_id": resource_id, "confirm": confirm})

在 apiapp/urls.py 中配置路由

apiapp/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('get_example/<int:user_id>/', views.get_example, name='get_example'),
    path('post_example/', views.post_example, name='post_example'),
    path('put_example/<int:item_id>/', views.put_example, name='put_example'),
    path('delete_example/', views.delete_example, name='delete_example'),
    path('api/resource/<str:resource_id>/', views.ApiResourceView.as_view(), name='api_resource'),
]

在 myproject/urls.py 中包含 apiapp.urls

myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('apiapp.urls')),
]

运行项目

python manage.py runserver