Airtable OAuth MCP 服务器是一个具备安全 OAuth 2.0 认证的模型上下文协议 (MCP) 服务器,可用于生产环境。该服务器使 AI 助手和应用程序能够通过标准化的 MCP 接口与 Airtable 数据库进行交互,为所有 Airtable 操作提供完整的 API 覆盖。
克隆仓库并安装依赖项:
git clone https://github.com/onimsha/airtable-mcp-server-oauth.git
cd airtable-mcp-server-oauth
uv sync
Client ID 和 Client Secret。http://localhost:8000/oauth/callback。复制环境模板并配置凭证:
cp .env.example .env
编辑 .env 文件并填入相应的值:
# Airtable OAuth 配置
AIRTABLE_CLIENT_ID="your_airtable_client_id_here"
AIRTABLE_CLIENT_SECRET="your_airtable_client_secret_here"
AIRTABLE_REDIRECT_URI="http://localhost:8000/oauth/callback"
# 服务器配置
HOST="0.0.0.0"
PORT=8000
LOG_LEVEL="INFO"
使用官方 MCP 检查器来测试并与服务器进行交互:
uv run python -m airtable_mcp http
http://localhost:8000/mcp。STDIO 传输(默认):
uv run python -m airtable_mcp
# 或者
uv run airtable-oauth-mcp
HTTP 传输:
uv run python -m airtable_mcp http
# 或者使用自定义主机/端口
uv run python -m airtable_mcp http localhost 8001
其他选项:
# 设置日志级别
uv run python -m airtable_mcp --log-level DEBUG
# 显示帮助信息
uv run python -m airtable_mcp --help
# 显示版本信息
uv run python -m airtable_mcp --version
HTTP 服务器将在 http://localhost:8000/(或自定义主机:端口)上可用,并带有用于 Web 集成的 OAuth 端点。
# 列出表中的所有记录
records = await client.call_tool("list_records", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY"
})
# 创建一条新记录
new_record = await client.call_tool("create_record", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY",
"fields": {
"Name": "John Doe",
"Email": "john@example.com",
"Status": "Active"
}
})
# 过滤搜索记录
filtered_records = await client.call_tool("search_records", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY",
"filter_by_formula": "AND({Status} = 'Active', {Email} != '')",
"fields": ["Name", "Email", "Status"]
})
# 带排序和过滤条件列出记录
records = await client.call_tool("list_records", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY",
"view": "Grid view",
"filter_by_formula": "{Priority} = 'High'",
"sort": [
{"field": "Created", "direction": "desc"},
{"field": "Name", "direction": "asc"}
],
"fields": ["Name", "Priority", "Created", "Status"]
})
# 批量操作
batch_create = await client.call_tool("create_records", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY",
"records": [
{"fields": {"Name": "Record 1", "Value": 100}},
{"fields": {"Name": "Record 2", "Value": 200}},
{"fields": {"Name": "Record 3", "Value": 300}}
],
"typecast": True
})
# 列出所有可访问的数据库
bases = await client.call_tool("list_bases")
# 获取特定表的详细信息
table_info = await client.call_tool("describe_table", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY"
})
# 列出数据库中的所有表
tables = await client.call_tool("list_tables", {
"base_id": "appXXXXXXXXXXXXXX",
"detail_level": "full"
})
服务器为 Airtable 操作提供了 10 个 MCP 工具:
数据库操作:
list_bases() - 列出所有可访问的数据库。list_tables(base_id, detail_level?) - 列出数据库中的表。describe_table(base_id, table_id) - 获取表的详细模式。记录操作:
list_records(base_id, table_id, view?, filter_by_formula?, sort?, fields?) - 带过滤条件列出记录。get_record(base_id, table_id, record_id) - 获取特定记录。create_record(base_id, table_id, fields, typecast?) - 创建单条记录。create_records(base_id, table_id, records, typecast?) - 创建多条记录。update_records(base_id, table_id, records, typecast?) - 更新多条记录。delete_records(base_id, table_id, record_ids) - 删除多条记录。search_records(base_id, table_id, filter_by_formula, view?, fields?) - 使用公式搜索记录。所有工具现在使用类型化参数而非通用的 args,这使得它们对 MCP 客户端更加透明。
参数灵活性:
fields 参数可以接受单个字段名(字符串)或字段名数组。sort 参数期望一个对象数组:[{"field": "Name", "direction": "asc"}]。git clone https://github.com/onimsha/airtable-mcp-server-oauth.git
cd airtable-mcp-server-oauth
uv sync --all-extras
uv run pytest
uv run pytest --cov=src/airtable_mcp --cov-report=html
类型检查:
uv run mypy src/
代码检查:
uv run ruff check src/
uv run ruff format src/
预提交钩子:
pip install pre-commit
pre-commit install
项目包含全面的测试覆盖:
# 运行所有测试
uv run pytest
# 运行带覆盖率的测试
uv run pytest --cov=src/airtable_mcp
# 运行特定的测试文件
uv run pytest tests/test_oauth.py
uv run pytest tests/test_tools.py
src/
├── airtable_mcp/ # 主要的 MCP 服务器包
│ ├── __init__.py # 包初始化
│ ├── __main__.py # 模块入口点
│ ├── main.py # CLI 和应用程序入口
│ ├── api/ # Airtable API 客户端
│ │ ├── __init__.py
│ │ ├── client.py # Airtable API 的 HTTP 客户端
│ │ ├── exceptions.py # API 特定的异常
│ │ └── models.py # API 响应的 Pydantic 模型
│ └── mcp/ # MCP 服务器实现
│ ├── __init__.py
│ ├── schemas.py # MCP 工具模式
│ └── server.py # 带有工具的 FastMCP 服务器
└── mcp_oauth_lib/ # 可重用的 OAuth 库
├── __init__.py # 库初始化
├── auth/ # 认证组件
│ ├── __init__.py
│ ├── context.py # 认证上下文管理
│ ├── middleware.py # OAuth 中间件
│ └── utils.py # 认证实用工具
├── core/ # 核心 OAuth 功能
│ ├── __init__.py
│ ├── config.py # OAuth 配置
│ ├── flow.py # OAuth 流程实现
│ └── server.py # OAuth 服务器端点
├── providers/ # OAuth 提供商实现
│ ├── __init__.py
│ ├── airtable.py # Airtable OAuth 提供商
│ └── base.py # 基础提供商接口
└── utils/ # OAuth 实用工具
├── __init__.py
├── pkce.py # PKCE 实现
└── state.py # 状态管理
所有配置通过环境变量(从 .env 文件加载)进行处理:
AIRTABLE_CLIENT_ID - 来自 Airtable 的 OAuth 客户端 ID。AIRTABLE_CLIENT_SECRET - OAuth 客户端密钥。AIRTABLE_REDIRECT_URI - OAuth 回调 URL。HOST - 服务器主机(默认:0.0.0.0)。PORT - 服务器端口(默认:8000)。LOG_LEVEL - 日志级别(默认:INFO)。MCP_SERVER_NAME - 服务器名称(可选)。MCP_SERVER_VERSION - 服务器版本(可选)。我们欢迎贡献!请遵循以下贡献指南:
本项目采用 MIT 许可证 - 详情请参阅 LICENSE 文件。