Airtable OAuth MCP服务器
  
生产准备就绪 模型上下文协议(MCP)服务器 用于具有安全OAuth 2.0身份验证的Airtable。该服务器使人工智能助手和应用程序能够通过标准化的MCP接口与Airtable基地进行交互,为所有Airtable操作提供完整的API覆盖范围。
🚀 特性
核心功能
- 🔐 OAuth 2.0身份验证 -使用Airtable进行基于令牌的安全身份验证
- 📊 完整的API Airtable覆盖范围 -10个全面的MCP工具,涵盖所有操作
- ⚡ FastMCP框架 -基于高性能FastMCP框架构建
- ☁️ 云就绪 -生产就绪部署支持
- 🔄 双重运输 -支持STDIO和HTTP传输协议
安全性和可靠性
- 🔑 基于环境的配置 -安全的凭证管理
- ✅ 类型安全 -Pydantic的完整类型提示和验证
- 🧪 综合测试 -使用pytest和覆盖率报告的单元测试
- 📝 代码质量 -用Ruff翻找和用MyPy检查类型
开发者体验
- 📚 丰富的文档 -全面的设置和使用指南
- 🔧 轻松设置 -使用uv包管理器进行简单安装
- 🎯 类型化参数 -清晰的键入工具参数,以获得更好的IDE支持
- 🔍 灵活查询 -高级过滤、排序和搜索功能
📋 先决条件
🚀 快速开始
1.安装
克隆存储库并安装依赖项:
git clone https://github.com/onimsha/airtable-mcp-server-oauth.git
cd airtable-mcp-server-oauth
uv sync2.Airtable OAuth设置
- 创建Airtable OAuth应用程序:
- 访问 Airtable开发者中心 - 创建新的OAuth集成 - 注意你的 Client ID 和 Client Secret - 将重定向URI设置为 http://localhost:8000/oauth/callback
3.环境配置
复制环境模板并配置凭据:
cp .env.example .env编辑 .env 与你的价值观:
# Airtable OAuth Configuration
AIRTABLE_CLIENT_ID="your_airtable_client_id_here"
AIRTABLE_CLIENT_SECRET="your_airtable_client_secret_here"
AIRTABLE_REDIRECT_URI="http://localhost:8000/oauth/callback"
# Server Configuration
HOST="0.0.0.0"
PORT=8000
LOG_LEVEL="INFO"4.使用MCP检查员进行测试
使用官方的MCP检查器测试并与您的服务器交互:
- 启动服务器:
uv run python -m airtable_mcp http- 打开MCP检查器:
访问 https://modelcontextprotocol.io/docs/tools/inspector
- 连接到您的服务器:
- 选择“HTTP流”传输 - 输入URL: http://localhost:8000/mcp - 点击“连接”
- 使用Airtable进行身份验证:
- 服务器将引导您完成OAuth身份验证 - 使用检查员测试可用的MCP工具
5.运行服务器
STDIO传输(默认):
uv run python -m airtable_mcp
# or
uv run airtable-oauth-mcpHTTP传输:
uv run python -m airtable_mcp http
# or with custom host/port
uv run python -m airtable_mcp http localhost 8001其他选项:
# Set log level
uv run python -m airtable_mcp --log-level DEBUG
# Show help
uv run python -m airtable_mcp --help
# Show version
uv run python -m airtable_mcp --versionHTTP服务器将在 http://localhost:8000/ (或自定义主机:端口)与OAuth端点进行web集成。
MCP工具可用
服务器为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"}]
💡 使用示例
基本记录操作
# List all records in a table
records = await client.call_tool("list_records", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY"
})
# Create a new record
new_record = await client.call_tool("create_record", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY",
"fields": {
"Name": "John Doe",
"Email": "john@example.com",
"Status": "Active"
}
})
# Search records with filtering
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"]
})高级查询
# List records with sorting and filtering
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 operations
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
})架构发现
# List all bases you have access to
bases = await client.call_tool("list_bases")
# Get detailed information about a specific table
table_info = await client.call_tool("describe_table", {
"base_id": "appXXXXXXXXXXXXXX",
"table_id": "tblYYYYYYYYYYYYYY"
})
# List all tables in a base
tables = await client.call_tool("list_tables", {
"base_id": "appXXXXXXXXXXXXXX",
"detail_level": "full"
})🛠️ 发展
入门指南
- 分叉和克隆:
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/Linting:
uv run ruff check src/
uv run ruff format src/预提交钩子:
pip install pre-commit
pre-commit install测试
该项目包括全面的测试覆盖范围:
- 单元测试: 测试单个组件和功能
- 集成测试: 测试OAuth流和Airtable API交互
- 覆盖范围报告: 确保代码覆盖率>90%
# Run all tests
uv run pytest
# Run with coverage
uv run pytest --cov=src/airtable_mcp
# Run specific test files
uv run pytest tests/test_oauth.py
uv run pytest tests/test_tools.py项目结构
src/
├── airtable_mcp/ # Main MCP server package
│ ├── __init__.py # Package initialization
│ ├── __main__.py # Module entry point
│ ├── main.py # CLI and application entry
│ ├── api/ # Airtable API client
│ │ ├── __init__.py
│ │ ├── client.py # HTTP client for Airtable API
│ │ ├── exceptions.py # API-specific exceptions
│ │ └── models.py # Pydantic models for API responses
│ └── mcp/ # MCP server implementation
│ ├── __init__.py
│ ├── schemas.py # MCP tool schemas
│ └── server.py # FastMCP server with tools
└── mcp_oauth_lib/ # Reusable OAuth library
├── __init__.py # Library initialization
├── auth/ # Authentication components
│ ├── __init__.py
│ ├── context.py # Auth context management
│ ├── middleware.py # OAuth middleware
│ └── utils.py # Auth utilities
├── core/ # Core OAuth functionality
│ ├── __init__.py
│ ├── config.py # OAuth configuration
│ ├── flow.py # OAuth flow implementation
│ └── server.py # OAuth server endpoints
├── providers/ # OAuth provider implementations
│ ├── __init__.py
│ ├── airtable.py # Airtable OAuth provider
│ └── base.py # Base provider interface
└── utils/ # OAuth utilities
├── __init__.py
├── pkce.py # PKCE implementation
└── state.py # State management⚙️ 配置
所有配置都通过环境变量(从加载 .env):
必需变量
AIRTABLE_CLIENT_ID-来自Airtable的OAuth客户端IDAIRTABLE_CLIENT_SECRET-OAuth客户端密钥AIRTABLE_REDIRECT_URI-OAuth回调URL
可选变量
HOST-服务器主机(默认值:0.0.0.0)PORT-服务器端口(默认值:8000)LOG_LEVEL-日志记录级别(默认值:INFO)MCP_SERVER_NAME-服务器名称(可选)MCP_SERVER_VERSION-服务器版本(可选)
🤝 贡献
我们欢迎捐款!请参阅我们的贡献指南:
- 复刻仓库 并创建一个特征分支
- 编写测试 对于任何新功能
- 确保代码质量 使用我们的linting和格式化工具
- 更新文档 对于任何API更改
- 提交拉取请求 描述清晰
贡献领域
- 🐛 错误修正 -帮助我们消灭虫子
- ✨ 新功能 -添加新的Airtable API端点
- 📚 文档 -改进设置指南和示例
- 🧪 测试 -提高测试覆盖率
- 🚀 演出 -优化API调用和缓存
📄 许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
🙏 致谢
📚 文档
额外资源
📞 支持
- 问题:
- 讨论:
- 文档: 维基工程
