通用代理连接器
      
面向AI代理的具有本体驱动语义路由的MCP基础设施
AI代理的连接层：安全的数据库访问、细粒度权限、自然语言查询和MCP治理。
💬 支持
- **** -问题和讨论
- **** -Bug报告和功能请求
- 文档 -技术指南和API参考
📚 额外资源
🤝 贡献
我们欢迎捐款！看 贡献.md 了解指导方针、代码风格以及如何提交pull请求。
快速链接：
特性
- 交互式演示项目：预构建的电子商务、SaaS指标和财务报告演示-请参阅 demos/README.md
- 代理商注册：使用唯一标识符注册和管理AI代理
- 认证：代理基于API密钥的安全身份验证
- 访问控制：细粒度权限管理系统
- RESTful API：用于代理管理的干净REST API
- 数据库就绪：可扩展的数据库连接器架构
- 安全凭据：使用Fernet对称加密对静态数据库凭据进行加密
- 多数据库支持：PostgreSQL、MySQL、MongoDB、BigQuery、Snowflake
- 连接池：可配置的连接池用于性能优化
- 超时管理：连接和查询的可配置超时
- AI代理管理：注册和管理多个AI代理提供商（OpenAI、Anthropic、本地模型、自定义模型）
- 气隙模式：通过本地人工智能模型支持完成网络隔离-无需外部API调用，数据永远不会离开您的网络
- 速率限制：每个代理的速率限制（每分钟/小时/天的查询数），以控制成本和资源使用
- 重试策略：失败代理请求的可配置重试策略（指数退避、固定延迟、线性）
- 版本控制：跟踪和回滚代理配置更改
- Webhook通知：代理查询成功/失败事件的实时通知
- 清除错误消息：用户友好、可操作的查询失败错误消息
- 数据库故障转移：当主数据库不可用时，自动故障转移到备份数据库
- 死信队列：修复问题后捕获并回放失败的查询
- 可视化：根据查询结果生成图表（条形图、直线图、饼图、散点图、面积图、表格、热图）
- 预定查询：安排定期查询（每小时、每天、每周、每月、自定义cron）
- 导出到外部系统：将结果导出到S3、谷歌表格、Slack、电子邮件、CSV、JSON、Excel
- 自然语言解释：查询结果的简明语言解释，包括统计数据和趋势
- A/B测试：在同一查询上测试不同的AI模型以比较性能
- 数据驻留规则：执行数据驻留规则（例如，欧盟数据保留在欧盟数据库中）以符合GDPR
- 数据保留策略：设置查询日志的保留策略，并自动清除
- 审计日志匿名化：在审计日志中匿名用户身份，同时保持问责制
- 上下文帮助工具提示：架构感知帮助工具提示，解释数据库、表和列
- 自动补全建议：自然语言查询中表/列名的自动补全
- 安装向导：引导安装向导，用于在5分钟内连接第一个数据库和代理
- 插件SDK：可扩展的插件系统，用于添加具有TypeScript类型和验证的自定义数据库驱动程序
- 可嵌入查询小部件：使用iframe嵌入代码、可自定义主题和安全的API密钥管理，将实时交互式查询窗口小部件添加到您的博客或网站-请参阅 WIDGE_EMBED_GUIDE.md
- CLI工具：用于在脚本和CI/CD中查询数据库的命令行界面-
npm install -g aidb-看 cli/README.md - Prompt工程工作室：用于使用变量、A/B测试和模板库自定义SQL生成提示的可视化编辑器-请参阅 docs/PROMPT_STUDIO_GUIDE.md
- 查询优化：使用EXPLAIN分析、索引推荐、查询重写和前后指标进行自动查询优化-请参阅 docs/QUERY_OPTIMIZATION_GUIDE.md
- 多Agent协作：协调多个代理，通过跟踪可视化在复杂查询（模式研究、SQL生成、验证）上进行协作-请参阅 docs/MULTI_GENT_COLLABORATION_GUIDE.md
- SSO集成：企业SSO支持SAML 2.0、OAuth 2.0和LDAP身份验证，包括属性映射-请参阅 docs/SSO_INTEGRATION-GUIDE.md
- 法律文件生成器：使用可定制的模板和多司法管辖区合规性（GDPR、CCPA、PIPEDA等）生成服务条款和隐私政策文件-请参阅 文档/法律_文档\_ GUIDE.md
- 退单报告：通过灵活的分配规则和发票生成，跟踪使用情况并按团队/用户分配成本-请参阅 docs/CHARGEBACK_GUIDE.md
- 采用分析：使用选择性匿名遥测、交互式仪表板和BI工具导出跟踪DAU、查询模式和功能使用情况-请参阅 docs/ADOPTION_ANALYTICS_GUIDE.md
- 培训数据导出：导出查询SQL对，以通过隐私安全匿名化、格式转换（JSONL/JSON/CSV）和数据集统计来微调自定义模型-请参阅 docs/TRAINING_DATA_EXPORT_GUIDE.md
项目结构
ai_agent_connector/
│
├── app/ # Core application logic
│ ├── __init__.py
│ ├── db/ # Database connection, models
│ │ ├── __init__.py
│ │ ├── connector.py # Database connector classes/functions
│ ├── agents/ # AI agent registration/auth
│ │ ├── __init__.py
│ │ ├── registry.py
│ ├── permissions/ # Access control logic
│ │ ├── __init__.py
│ │ ├── access_control.py
│ ├── api/ # API endpoints/routes
│ │ ├── __init__.py
│ │ ├── routes.py
│ └── utils/ # Utility functions/helpers
│ ├── __init__.py
│ └── helpers.py
│
├── tests/ # Unit tests
│
├── config/ # Configuration files (env, secrets)
│
├── requirements.txt # Python dependencies
├── README.md
└── main.py # Entry point安装
- 创建虚拟环境：
python -m venv venv- 激活虚拟环境：
# Windows
.\venv\Scripts\Activate.ps1
# Linux/Mac
source venv/bin/activate- 安装依赖项：
pip install -r requirements.txt- 设置加密密钥（用于生产）：
# Generate a secure encryption key
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
# Set as environment variable
export ENCRYPTION_KEY="your-generated-key-here"备注：加密密钥是保护数据库凭据所必需的。如果没有设置，将生成一个临时密钥（不适合生产）。
用法
运行应用程序
python main.py应用程序将于启动 http://127.0.0.1:5000 默认情况下。
交互式演示项目
尝试使用示例数据进行交互式演示：
- 电子商务分析演示 -分析销售、客户和产品
- SaaS指标仪表板演示 -跟踪MRR、流失率和用户增长
- 财务报告演示 -生成财务报告并分析交易
每个演示包括：
- ✅ 具有真实数据的示例数据库
- ✅ 逐步演练（2分钟）
- ✅ 自然语言查询示例
- ✅ 即用型代理配置
快速入门：
# Setup all demos at once
./demos/setup_all_demos.sh # Linux/Mac
demos\setup_all_demos.ps1 # Windows
# Or setup individual demos
createdb ecommerce_demo
psql -U postgres -d ecommerce_demo -f demos/ecommerce/setup.sql看 demos/README.md 获取完整的演示文档。
Web仪表板
对于非技术用户，可以在以下网址获得一个简单的基于网络的仪表板：
- 仪表盘:
http://127.0.0.1:5000/dashboard-代理和系统状态概述 - 集成向导:
http://127.0.0.1:5000/wizard-连接代理和数据库的分步指南 - 代理商管理:
http://127.0.0.1:5000/agents-查看和管理所有已注册的代理 - 访问预览:
http://127.0.0.1:5000/agents//access-preview-预览代理可以访问哪些表/字段
仪表板提供了一个用户友好的界面，用于：
- 无需编写代码即可将新代理连接到数据库
- 注册前测试数据库连接
- 查看和管理现有代理
- 通过向导设置权限
- 预览代理对表和字段的访问权限（自助服务权限透明度）
按照向导步骤执行以下操作：
- 输入代理信息
- 配置数据库连接（带连接测试）
- 提供代理凭据
- 查看并连接
仪表板在后台自动处理所有API调用。
环境变量
您可以使用环境变量配置应用程序：
FLASK_ENV：环境模式（开发、生产、测试）PORT：服务器端口（默认值：5000）HOST：服务器主机（默认值：127.0.0.1）SECRET_KEY：会话的Flask密钥DATABASE_URL：数据库连接字符串OPENAI_API_KEY：用于自然语言查询转换的OpenAI API密钥（NL查询需要，气隙模式不需要）AIR_GAPPED_MODE：启用气隙模式以阻止外部API调用（默认值：false）LOCAL_AI_BASE_URL：本地AI模型API的基本URL（默认值：http://localhost:11434Ollama）LOCAL_AI_MODEL：要使用的默认本地AI模型（默认值：llama2）
API终点
健康检查
GET /api/health测试数据库连接
POST /api/databases/test
Content-Type: application/json
{
"connection_string": "postgresql://db_user:db_pass@db.example.com/analytics"
}在注册代理之前测试数据库连接。可用于以编程方式验证凭据。
替代格式（单个参数）：
{
"host": "db.example.com",
"port": 5432,
"user": "db_user",
"password": "db_pass",
"database": "analytics"
}响应（成功）：
{
"status": "success",
"message": "Database connection test successful",
"database_info": {
"connection_string": "***",
"connection_name": "default",
"type": "postgresql"
}
}响应（失败）：
{
"status": "error",
"message": "Database connection failed: connection refused",
"error": "connection refused"
}注册代理人
POST /api/agents/register
Content-Type: application/json
{
"agent_id": "agent-001",
"agent_info": {
"name": "Reporting Agent",
"type": "assistant"
},
"agent_credentials": {
"api_key": "agent-issued-key",
"api_secret": "agent-issued-secret"
},
"database": {
"connection_string": "postgresql://db_user:db_pass@db.example.com/analytics",
"connection_name": "analytics",
"type": "postgresql"
}
}注册流程：
- 安全地对提供的代理凭据进行哈希运算
- 通过执行连接测试来验证提供的数据库详细信息
- 将代理链接到已验证的数据库，以便将来的查询可以通过连接器路由
- 返回代理用于身份验证的API密钥
答复：
{
"agent_id": "agent-001",
"api_key": "generated-api-key-here",
"database": {
"status": "connected",
"connection_name": "analytics",
"type": "postgresql"
},
"message": "Agent registered with database connectivity"
}更新代理数据库连接
PUT /api/agents//database
Content-Type: application/json
{
"connection_string": "postgresql://new_user:new_pass@new_host:5432/new_db"
}更新或添加现有代理的数据库连接。适用于：
- 更改数据库凭据
- 切换到其他数据库
- 向没有数据库连接的代理添加数据库连接
答复：
{
"message": "Database connection updated for agent agent-001",
"agent_id": "agent-001",
"database": {
"status": "connected",
"connection_name": "default",
"type": "postgresql"
},
"updated_at": "2024-01-15T10:30:00Z"
}列出可用表/数据集
GET /api/agents//tables列出代理连接的数据库中的所有可用表和数据集。这有助于管理员查看他们可以对哪些资源设置权限。响应包括每个表的权限信息。
答复：
{
"agent_id": "agent-001",
"database": "analytics",
"tables": [
{
"schema": "public",
"table_name": "users",
"resource_id": "users",
"type": "table",
"permissions": ["read", "write"],
"has_read": true,
"has_write": true
},
{
"schema": "analytics",
"table_name": "sales",
"resource_id": "analytics.sales",
"type": "table",
"permissions": ["read"],
"has_read": true,
"has_write": false
}
],
"count": 2
}预览代理访问（自助服务）
GET /api/agents//access-preview提供代理可以访问的表和字段的全面预览，使自助服务用户的权限透明。此端点显示：
- 摘要统计（表总数、可访问/不可访问计数、权限细分）
- 可访问表的详细信息，包括列级详细信息
- 无法访问的表列表
答复：
{
"agent_id": "agent-001",
"database": "analytics",
"summary": {
"total_tables": 5,
"accessible_tables": 3,
"inaccessible_tables": 2,
"read_only_tables": 1,
"read_write_tables": 2,
"write_only_tables": 0
},
"accessible_tables": [
{
"schema": "public",
"table_name": "users",
"resource_id": "users",
"access_status": "accessible",
"permissions": ["read", "write"],
"has_read": true,
"has_write": true,
"column_count": 3,
"columns": [
{
"name": "id",
"type": "integer",
"nullable": false,
"default": null,
"position": 1
},
{
"name": "name",
"type": "varchar",
"nullable": true,
"default": null,
"position": 2
},
{
"name": "email",
"type": "varchar",
"nullable": true,
"default": null,
"position": 3
}
]
},
{
"schema": "public",
"table_name": "orders",
"resource_id": "orders",
"access_status": "accessible",
"permissions": ["read"],
"has_read": true,
"has_write": false,
"column_count": 2,
"columns": [
{
"name": "id",
"type": "integer",
"nullable": false,
"default": null,
"position": 1
},
{
"name": "user_id",
"type": "integer",
"nullable": true,
"default": null,
"position": 2
}
]
}
],
"inaccessible_tables": [
{
"schema": "analytics",
"table_name": "sales",
"resource_id": "analytics.sales",
"access_status": "no_permission",
"permissions": [],
"has_read": false,
"has_write": false,
"column_count": 0,
"columns": []
}
]
}
}Web界面： 访问视觉预览： http://127.0.0.1:5000/agents//access-preview
web界面提供：
- 显示权限统计信息的可视化摘要卡
- 可访问表的可扩展列详细信息
- 明确区分可访问表和不可访问表
- 权限徽章（读/写）便于识别
设置表/数据集权限
PUT /api/agents//permissions/resources
Content-Type: application/json
{
"resource_id": "public.sales_orders",
"resource_type": "table", # or "dataset"
"permissions": ["read", "write"]
}使用此端点管理已注册代理的细粒度读/写访问。伴侣 GET /api/agents//permissions/resources call列出了当前的资源级别权限。
权限类型：
read：SELECT查询需要write：INSERT、UPDATE、DELETE查询需要delete：供将来使用（目前未强制执行）admin：用于行政操作（目前尚未执行）
列出资源权限
GET /api/agents//permissions/resources列出当前授予代理的所有资源级权限。
答复：
{
"agent_id": "agent-001",
"resources": {
"public.orders": {
"type": "table",
"permissions": ["read", "write"]
},
"analytics.sales": {
"type": "table",
"permissions": ["read"]
}
}
}撤销资源权限
DELETE /api/agents//permissions/resources/撤销代理对特定资源（表或数据集）的所有权限。
答复：
{
"message": "Permissions revoked for resource public.orders",
"agent_id": "agent-001",
"resource_id": "public.orders"
}使用权限强制执行查询
POST /api/agents//query
Content-Type: application/json
X-API-Key:
{
"query": "SELECT * FROM public.users WHERE id = %s",
"params": [1], # Optional: query parameters
"as_dict": false # Optional: return results as dictionaries
}此终结点执行具有自动权限强制的数据库查询：
特征：
- 自动权限验证：检查代理是否对查询访问的所有表/数据集具有所需权限
- 查询类型检测：自动确定查询是否需要读取或写入权限
- 表格提取：解析SQL以识别正在访问的所有表/数据集
- 安全执行：仅在代理具有适当权限时执行查询
响应（成功）：
{
"agent_id": "agent-001",
"query_type": "SELECT",
"tables_accessed": ["public.users"],
"success": true,
"result": [["user1"], ["user2"]],
"row_count": 2
}响应（权限被拒绝）：
{
"error": "Permission denied",
"denied_resources": [
{
"resource": "public.orders",
"required_permission": "read",
"message": "Agent does not have read permission on public.orders"
}
],
"message": "Agent lacks required permissions on one or more resources"
}支持的查询类型：
SELECT：需要read许可INSERT：需要write许可UPDATE：需要write许可DELETE：需要write许可
工作流程示例：
- 注册具有数据库连接的代理
- 设置特定表/数据集的权限
- 代理使用其API密钥执行查询
- 系统在执行前自动验证权限
自然语言查询
POST /api/agents//query/natural
Content-Type: application/json
X-API-Key:
{
"query": "Show me all users who are older than 25",
"as_dict": false # Optional: return results as dictionaries
}此端点允许管理员用简单的英语提交问题。系统自动：
- 将自然语言问题转换为SQL
- 验证所需表的权限
- 执行查询
- 返回结果
特征：
- 自动生成SQL：使用AI将自然语言转换为SQL
- 模式感知：自动包含数据库架构信息，以便准确生成SQL
- 许可执行：权限检查与直接SQL查询相同
- 错误处理：如果转换或执行失败，则提供详细的错误消息
响应（成功）：
{
"agent_id": "agent-001",
"natural_language_query": "Show me all users who are older than 25",
"generated_sql": "SELECT * FROM users WHERE age > 25",
"query_type": "SELECT",
"tables_accessed": ["users"],
"success": true,
"result": [["user1", 30], ["user2", 28]],
"row_count": 2
}响应（权限被拒绝）：
{
"error": "Permission denied",
"denied_resources": [
{
"resource": "users",
"required_permission": "read",
"message": "Agent does not have read permission on users"
}
],
"generated_sql": "SELECT * FROM users WHERE age > 25",
"natural_language_query": "Show me all users who are older than 25"
}配置：
- 集
OPENAI_API_KEY启用自然语言查询的环境变量 - 用途
gpt-4o-mini默认模型（可配置）
列出代理
GET /api/agents获取代理
GET /api/agents/撤销代理
DELETE /api/agents/完全撤销代理对系统的访问权限。此操作：
删除的内容：
- 代理注册和元数据
- 所有API密钥（代理无法再进行身份验证）
- 所有权限（常规和资源级别）
- 数据库连接配置
- 存储的凭据
安全：
- 撤销后，代理无法进行身份验证或访问任何资源
- 所有访问立即失效
- 出于审计目的，会记录撤销情况
答复：
{
"message": "Agent agent-001 revoked successfully",
"details": {
"agent_id": "agent-001",
"permissions_revoked": true,
"api_keys_invalidated": true,
"database_access_removed": true,
"credentials_removed": true
}
}使用案例：
- 不再需要代理
- 安全问题或凭据泄露
- 代理商变更或更换
- 移除访问权限的合规要求
AI代理管理端点
所有AI代理管理端点都需要管理员权限。使用 X-API-Key 具有管理代理的API密钥的头。
注册AI代理
POST /api/admin/ai-agents/register
Content-Type: application/json
X-API-Key:
{
"agent_id": "openai-agent-1",
"provider": "openai",
"model": "gpt-4",
"api_key": "sk-...",
"temperature": 0.7,
"max_tokens": 2000,
"rate_limit": {
"queries_per_minute": 60,
"queries_per_hour": 1000
},
"retry_policy": {
"enabled": true,
"max_retries": 3,
"strategy": "exponential",
"initial_delay": 1.0
}
}注册一个支持以下功能的AI代理：
- 提供商:
openai,anthropic,local（适用于本地AI模型），或custom - 气隙模式：启用时，仅
local允许提供者。看 AIR_GAPPED_MODE.md 了解详情。 - 速率限制：控制每分钟/每小时/每天的查询数
- 重试策略：为失败的请求配置重试策略
- 版本控制：配置更改的自动版本跟踪
答复：
{
"agent_id": "openai-agent-1",
"provider": "openai",
"model": "gpt-4",
"version": 1,
"registered_at": "2024-01-15T10:30:00Z"
}列出AI代理
GET /api/admin/ai-agents
X-API-Key: 答复：
{
"agents": [
{
"agent_id": "openai-agent-1",
"configuration": {
"provider": "openai",
"model": "gpt-4",
"temperature": 0.7
},
"rate_limit": {
"queries_per_minute": 60,
"queries_per_hour": 1000
},
"retry_policy": {
"enabled": true,
"max_retries": 3,
"strategy": "exponential"
},
"current_version": 1
}
],
"count": 1
}执行查询
POST /api/admin/ai-agents//query
Content-Type: application/json
X-API-Key:
{
"query": "What is machine learning?",
"context": {
"system_prompt": "You are a helpful assistant."
}
}使用指定的AI代理执行查询。自动应用速率限制、重试策略并发送webhook通知。
答复：
{
"response": "Machine learning is...",
"model": "gpt-4",
"usage": {
"prompt_tokens": 10,
"completion_tokens": 50,
"total_tokens": 60
},
"provider": "openai"
}设置速率限制
POST /api/admin/ai-agents//rate-limit
Content-Type: application/json
X-API-Key:
{
"queries_per_minute": 100,
"queries_per_hour": 2000,
"queries_per_day": 10000
}配置速率限制以控制成本和资源使用。
获取速率限制使用情况
GET /api/admin/ai-agents//rate-limit
X-API-Key: 答复：
{
"agent_id": "openai-agent-1",
"rate_limit": {
"queries_per_minute": 100,
"queries_per_hour": 2000
},
"usage": {
"rate_limits_configured": true,
"limits": {
"queries_per_minute": 100,
"queries_per_hour": 2000
},
"current_usage": {
"queries_last_minute": 5,
"queries_last_hour": 50,
"queries_last_day": 500
},
"remaining": {
"queries_this_minute": 95,
"queries_this_hour": 1950,
"queries_this_day": 9500
}
}
}设置重试策略
POST /api/admin/ai-agents//retry-policy
Content-Type: application/json
X-API-Key:
{
"enabled": true,
"max_retries": 5,
"strategy": "exponential",
"initial_delay": 1.0,
"max_delay": 60.0,
"backoff_multiplier": 2.0,
"retryable_errors": ["timeout", "connection_error", "rate_limit"],
"jitter": true
}配置重试策略以处理瞬态错误。策略：
fixed：修复了重试之间的延迟exponential：指数回退（默认）linear：线性退避
列出配置版本
GET /api/admin/ai-agents//versions?limit=10
X-API-Key: 答复：
{
"agent_id": "openai-agent-1",
"versions": [
{
"version": 2,
"timestamp": "2024-01-15T11:00:00Z",
"config": {
"provider": "openai",
"model": "gpt-4",
"temperature": 0.8
},
"description": "Updated temperature",
"created_by": "admin"
},
{
"version": 1,
"timestamp": "2024-01-15T10:30:00Z",
"config": {
"provider": "openai",
"model": "gpt-4",
"temperature": 0.7
},
"description": "Initial configuration",
"created_by": "admin"
}
],
"count": 2
}回滚配置
POST /api/admin/ai-agents//rollback
Content-Type: application/json
X-API-Key:
{
"version": 1,
"description": "Rolling back due to issues"
}回滚到以前的配置版本。使用回滚配置创建新版本。
答复：
{
"agent_id": "openai-agent-1",
"rollback_to_version": 1,
"new_version": {
"version": 3,
"timestamp": "2024-01-15T12:00:00Z",
"config": {...},
"tags": ["rollback", "from_version_1"]
},
"message": "Configuration rolled back successfully"
}注册Webhook
POST /api/admin/ai-agents//webhooks
Content-Type: application/json
X-API-Key:
{
"url": "https://example.com/webhook",
"events": ["query_success", "query_failure", "rate_limit_exceeded"],
"secret": "webhook-secret",
"timeout": 10,
"retry_on_failure": true,
"max_retries": 3
}注册webhook以接收代理事件的通知：
query_success：查询已成功执行query_failure：重试后查询失败rate_limit_exceeded：已超出速率限制configuration_changed：配置已更新agent_registered：新代理已注册agent_revoked：代理已删除
答复：
{
"agent_id": "openai-agent-1",
"webhook_id": "webhook_1234567890",
"webhook": {
"url": "https://example.com/webhook",
"events": ["query_success", "query_failure"],
"secret": "***",
"timeout": 10,
"enabled": true
},
"message": "Webhook registered successfully"
}获取Webhook历史记录
GET /api/admin/ai-agents//webhooks/history?limit=100
X-API-Key: 答复：
{
"agent_id": "openai-agent-1",
"history": [
{
"webhook_url": "https://example.com/webhook",
"event": "query_success",
"timestamp": "2024-01-15T10:30:00Z",
"status": "success",
"response_code": 200,
"attempts": 1
}
],
"statistics": {
"total_deliveries": 100,
"successful": 95,
"failed": 5,
"success_rate": 95.0
}
}API文档
GET /api/api-docs返回与OpenAPI 3.0兼容的API文档。适用于：
- API客户端生成
- 理解请求/响应模式
- 与API文档工具集成
答复： OpenAPI 3.0 JSON规范
审计日志
GET /api/audit/logs使用筛选选项检索审核日志。所有查询和代理操作都会自动记录以供审计。
查询参数：
agent_id：按代理ID筛选action_type：按操作类型筛选（query_execution、natural_language_query、agent_registered、permission_set等）status：按状态筛选（成功、错误、拒绝）limit：要返回的最大日志数（默认值：100，最大值：1000）offset：分页时要跳过的日志数（默认值：0）
答复：
{
"logs": [
{
"id": 1,
"timestamp": "2024-01-15T10:30:00Z",
"action_type": "query_execution",
"agent_id": "agent-001",
"status": "success",
"details": {
"query_type": "SELECT",
"tables_accessed": ["users"],
"row_count": 5,
"query_preview": "SELECT * FROM users WHERE age > 25"
}
}
],
"total": 150,
"limit": 100,
"offset": 0,
"has_more": true
}动作类型：
query_execution：直接执行SQL查询natural_language_query：自然语言查询执行agent_registered：代理人注册agent_revoked：代理撤销permission_set：权限分配permission_revoked：权限撤销permission_listed：权限列表tables_listed：表格列表agent_viewed：代理信息检索agents_listed：代理列表检索
获取特定审核日志
GET /api/audit/logs/按ID检索特定的审核日志条目。
审计统计
GET /api/audit/statistics获取有关审核日志的统计信息。
查询参数：
agent_id：按代理ID筛选统计信息（可选）
答复：
{
"total_actions": 150,
"by_action_type": {
"query_execution": 80,
"natural_language_query": 20,
"agent_registered": 5,
"permission_set": 10
},
"by_status": {
"success": 140,
"error": 8,
"denied": 2
},
"recent_actions": [...]
}安全通知
GET /api/notifications获取安全通知和警报。系统会自动监控安全问题和异常访问模式。
查询参数：
severity：按严重程度过滤（低、中、高、严重）agent_id：按代理ID筛选unread_only：仅返回未读通知（true/false）limit：要返回的最大通知数（默认值：100，最大值：1000）
答复：
{
"notifications": [
{
"id": 1,
"timestamp": "2024-01-15T10:30:00Z",
"event_type": "failed_authentication",
"severity": "medium",
"agent_id": "agent-001",
"message": "Failed authentication attempt",
"details": {
"error": "Invalid API key",
"ip": "192.168.1.1"
},
"read": false
}
],
"total": 25,
"unread_count": 5,
"count": 25
}监控的安全事件：
- 身份验证失败：多次登录尝试失败
- 权限不足：试图访问未经授权的资源
- 多重故障：短时间内重复故障（异常）
- 异常访问模式：异常的查询率或模式
- 代理人已撤销：代理访问撤销事件
- 超出费率限制：查询率过高
严重级别：
critical：需要立即关注（例如，多个安全漏洞）high：重要的安全事件（例如，代理撤销、多次失败）medium：安全问题（例如，身份验证失败、权限被拒绝）low：信息安全事件
将通知标记为已读
PUT /api/notifications//read将特定通知标记为已读。
将所有通知标记为已读
PUT /api/notifications/read-all将所有通知标记为已读。
通知统计
GET /api/notifications/stats获取有关安全通知的统计信息。
答复：
{
"total": 25,
"unread": 5,
"by_severity": {
"critical": 2,
"high": 5,
"medium": 15,
"low": 3
},
"by_event_type": {
"failed_authentication": 10,
"permission_denied": 8,
"multiple_failures": 2
},
"recent_critical": [...]
}仪表板访问
web仪表板包括 安全警报 显示最近安全通知的部分。访问完整的通知页面 /notifications 查看所有警报、按严重性筛选和管理通知状态。
开发者指南
程序集成
API的设计是为了方便编程集成。以下是一个典型的工作流程：
1.测试数据库连接：
import requests
response = requests.post('http://localhost:5000/api/databases/test', json={
'connection_string': 'postgresql://user:pass@localhost/db'
})
if response.json()['status'] == 'success':
print("Database connection valid")2.注册代理人：
response = requests.post('http://localhost:5000/api/agents/register', json={
'agent_id': 'my-agent',
'agent_credentials': {
'api_key': 'agent-key',
'api_secret': 'agent-secret'
},
'database': {
'connection_string': 'postgresql://user:pass@localhost/db'
}
})
api_key = response.json()['api_key']3.设置权限：
requests.put(
f'http://localhost:5000/api/agents/my-agent/permissions/resources',
json={
'resource_id': 'users',
'permissions': ['read', 'write']
}
)4.执行查询：
response = requests.post(
'http://localhost:5000/api/agents/my-agent/query',
json={'query': 'SELECT * FROM users LIMIT 10'},
headers={'X-API-Key': api_key}
)
results = response.json()['result']5.更新数据库连接：
requests.put(
'http://localhost:5000/api/agents/my-agent/database',
json={
'connection_string': 'postgresql://new_user:new_pass@new_host/db'
}
)数据库故障转移端点
注册故障转移端点
POST /api/admin/agents//failover/endpoints
Content-Type: application/json
{
"endpoints": [
{
"name": "Primary Database",
"host": "db-primary.example.com",
"port": 5432,
"user": "user",
"password": "pass",
"database": "mydb",
"database_type": "postgresql",
"is_primary": true,
"priority": 0
},
{
"name": "Backup Database",
"host": "db-backup.example.com",
"port": 5432,
"user": "user",
"password": "pass",
"database": "mydb",
"database_type": "postgresql",
"is_primary": false,
"priority": 1
}
]
}获取故障切换状态
GET /api/admin/agents//failover/status答复：
{
"agent_id": "agent-001",
"status": "primary",
"current_endpoint": {
"endpoint_id": "...",
"name": "Primary Database",
"is_primary": true
},
"endpoints": [...],
"available_endpoints": 2,
"total_endpoints": 2
}重置端点
POST /api/admin/agents//failover/endpoints//reset死信队列端点
列出DLQ条目
GET /api/admin/dlq/entries?agent_id=&status=pending&limit=100获取DLQ条目
GET /api/admin/dlq/entries/重播DLQ条目
POST /api/admin/dlq/entries//replay答复：
{
"entry": {...},
"result": [...],
"message": "Query replayed successfully",
"retry_count": 1
}归档DLQ条目
POST /api/admin/dlq/entries//archive删除DLQ条目
DELETE /api/admin/dlq/entries/获取DLQ统计数据
GET /api/admin/dlq/statistics?agent_id=答复：
{
"total_entries": 10,
"status_counts": {
"pending": 5,
"success": 3,
"failed": 2
},
"error_type_counts": {
"ConnectionError": 5,
"SyntaxError": 3
}
}Clear Agent DLQ
POST /api/admin/dlq/agents//clear错误处理
所有端点都返回一致的错误响应。查询失败现在包括格式化的错误消息：
查询错误响应：
{
"error": "Query execution failed",
"user_friendly_message": "Invalid column name: 'invalid_column'. Please check the column name and try again.",
"error_type": "Exception",
"actionable_details": {
"column": "invalid_column"
},
"suggested_fixes": [
"Check the column name spelling and case sensitivity",
"Verify the column exists in the table"
],
"generated_sql": "SELECT invalid_column FROM users",
"dlq_entry_id": "...",
"failover_attempted": false
}{
"error": "Error type",
"message": "Detailed error message",
"agent_id": "agent-001" // if applicable
}常见HTTP状态代码：
200：成功201：已创建（代理已注册）400：错误请求（验证错误）401：未经授权（API密钥丢失/无效）403：禁止（拒绝许可）404：未找到（未找到代理/资源）500：内部服务器错误
API文档
访问OpenAPI规范 /api/api-docs 获取完整的API文档、请求/响应模式和集成详细信息。
发展
以开发模式运行
export FLASK_ENV=development
python main.py测试
测试位于 tests/ 目录。使用以下工具运行测试：
pytest tests/插件SDK
AI Agent Connector包含一个插件SDK，允许开发人员为专有或利基数据库创建自定义数据库驱动程序插件。
概述
插件SDK提供：
- 基础插件类:
DatabasePlugin-所有插件的抽象基类 - 插件注册表：插件的自动注册和发现
- TypeScript类型：插件开发的类型定义
- 验证：内置配置验证
- 整合：与现有数据库连接器工厂无缝集成
创建插件
- 创建插件类
from ai_agent_connector.app.db.plugin import DatabasePlugin
from ai_agent_connector.app.db.base_connector import BaseDatabaseConnector
from typing import Dict, Any, List, Optional, Union, Tuple
class MyCustomConnector(BaseDatabaseConnector):
"""Your custom database connector implementation"""
def __init__(self, config: Dict[str, Any]):
super().__init__(config)
# Initialize your database client here
def connect(self) -> bool:
# Implement connection logic
self._is_connected = True
return True
def disconnect(self) -> None:
# Implement disconnection logic
self._is_connected = False
def execute_query(
self,
query: str,
params: Optional[Union[Dict[str, Any], Tuple, List]] = None,
fetch: bool = True,
as_dict: bool = False
) -> Optional[Union[List[Tuple], List[Dict[str, Any]]]]:
# Implement query execution
return [] if fetch else None
@property
def is_connected(self) -> bool:
return self._is_connected
def get_database_info(self) -> Dict[str, Any]:
return {'type': 'my_custom_db', 'version': '1.0.0'}
class MyCustomDatabasePlugin(DatabasePlugin):
"""Plugin for My Custom Database"""
@property
def plugin_name(self) -> str:
return "my_custom_db_plugin"
@property
def plugin_version(self) -> str:
return "1.0.0"
@property
def database_type(self) -> str:
return "my_custom_db"
@property
def display_name(self) -> str:
return "My Custom Database"
@property
def description(self) -> str:
return "Plugin for connecting to My Custom Database"
@property
def required_config_keys(self) -> List[str]:
return ['host', 'database', 'api_key']
@property
def optional_config_keys(self) -> List[str]:
return ['port', 'timeout']
def create_connector(self, config: Dict[str, Any]) -> BaseDatabaseConnector:
return MyCustomConnector(config)
def detect_database_type(self, config: Dict[str, Any]) -> Optional[str]:
if config.get('type') == 'my_custom_db':
return 'my_custom_db'
return None- 注册插件
from ai_agent_connector.app.db.plugin import register_plugin
plugin = MyCustomDatabasePlugin()
register_plugin(plugin)- 使用插件
from ai_agent_connector.app.db import DatabaseConnector
# Use your custom database type
connector = DatabaseConnector(
database_type='my_custom_db',
host='localhost',
database='mydb',
api_key='your-api-key'
)
connector.connect()
results = connector.execute_query("SELECT * FROM users")
connector.disconnect()从文件加载插件
您可以从Python文件加载插件：
from ai_agent_connector.app.db.plugin import get_plugin_registry
registry = get_plugin_registry()
# Load a single plugin
plugin = registry.load_plugin_from_file('/path/to/plugin.py')
# Load all plugins from a directory
plugins = registry.load_plugins_from_directory('/path/to/plugins')插件API终结点
插件SDK包括用于插件管理的REST API端点：
GET /api/plugins-列出所有已注册的插件GET /api/plugins/-获取插件信息DELETE /api/plugins/-注销插件POST /api/plugins/load-从文件加载插件POST /api/plugins/load-directory-从目录加载插件POST /api/plugins/validate-验证插件配置GET /api/plugins/supported-types-获取所有支持的数据库类型
插件示例
看 examples/plugins/example_custom_db.py 查看完整的插件实现示例。
TypeScript类型
TypeScript类型定义可在 ai_agent_connector/app/db/plugin_types.ts 用于开发插件或与TypeScript/JavaScript应用程序集成时的参考。
验证测试
插件SDK包括全面的验证测试。用以下方式运行它们：
pytest tests/test_plugin_sdk.py插件要求
所有插件必须：
- 扩展
DatabasePlugin基类 - 实现所有抽象方法和属性
- 返回延伸的连接器
BaseDatabaseConnector - 实施适当的错误处理
- 在创建连接器之前验证配置
许可证
MIT许可证
贡献
欢迎投稿！请随时提交拉取请求。