github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-08 03:00:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

feat(blocks): Add Airtable Integration with Base Management (#10485)

Browse Source 
### Changes 🏗️

This PR adds Airtable integration to AutoGPT with the following blocks:

- **List Bases Block**: Lists all Airtable bases accessible to the
authenticated user
- **Create Base Block**: Creates new Airtable bases with specified
workspace and name

<img width="1294" height="879" alt="Screenshot 2025-07-30 at 11 03 43"
src="https://github.com/user-attachments/assets/0729e2e8-b254-4ed6-9481-1c87a09fb1c8"
/>


### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] Tested create base block
- [x] Tested list base block
This commit is contained in:
Swifty
2025-07-30 13:03:07 +02:00
committed by GitHub
parent b08761816a
commit 02f5e92167
3 changed files with 228 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
autogpt_platform/backend/backend/blocks/airtable/_api.py
View File

@@ -1141,3 +1141,88 @@ async def oauth_refresh_tokens(
if response.ok:
return OAuthTokenResponse.model_validate(response.json())
raise ValueError(f"Failed to refresh tokens: {response.status} {response.text}")
#################################################################
# Base Management
#################################################################
async def create_base(
credentials: Credentials,
workspace_id: str,
name: str,
tables: list[dict] = [
{
"description": "Default table",
"name": "Default table",
"fields": [
{
"name": "ID",
"type": "number",
"description": "Auto-incrementing ID field",
"options": {"precision": 0},
}
],
}
],
) -> dict:
"""
Create a new base in Airtable.
Args:
credentials: Airtable API credentials
workspace_id: The workspace ID where the base will be created
name: The name of the new base
tables: Optional list of table objects to create in the base
Returns:
dict: Response containing the created base information
"""
params: dict[str, Any] = {
"name": name,
"workspaceId": workspace_id,
}
if tables:
params["tables"] = tables
print(params)
response = await Requests().post(
"https://api.airtable.com/v0/meta/bases",
headers={
"Authorization": credentials.auth_header(),
"Content-Type": "application/json",
},
json=params,
)
return response.json()
async def list_bases(
credentials: Credentials,
offset: str | None = None,
) -> dict:
"""
List all bases that the authenticated user has access to.
Args:
credentials: Airtable API credentials
offset: Optional pagination offset
Returns:
dict: Response containing the list of bases
"""
params = {}
if offset:
params["offset"] = offset
response = await Requests().get(
"https://api.airtable.com/v0/meta/bases",
headers={"Authorization": credentials.auth_header()},
params=params,
)
return response.json()

26
autogpt_platform/backend/backend/blocks/airtable/_api_test.py
View File

@@ -9,6 +9,7 @@ from ._api import (
TableFieldType,
WebhookFilters,
WebhookSpecification,
create_base,
create_field,
create_record,
create_table,
@@ -17,6 +18,7 @@ from ._api import (
delete_record,
delete_webhook,
get_record,
list_bases,
list_records,
list_webhook_payloads,
update_field,
@@ -38,7 +40,21 @@ async def test_create_update_table():
api_key=SecretStr(key),
)
postfix = uuid4().hex[:4]
base_id = "appSbaQLkcYiIOqux"
workspace_id = "wsphuHmfllg7V3Brd"
response = await create_base(credentials, workspace_id, "API Testing Base")
assert response is not None, f"Checking create base response: {response}"
assert (
response.get("id") is not None
), f"Checking create base response id: {response}"
base_id = response.get("id")
assert base_id is not None, f"Checking create base response id: {base_id}"
response = await list_bases(credentials)
assert response is not None, f"Checking list bases response: {response}"
assert "API Testing Base" in [
base.get("name") for base in response.get("bases", [])
], f"Checking list bases response bases: {response}"
table_name = f"test_table_{postfix}"
table_fields = [{"name": "test_field", "type": "singleLineText"}]
table = await create_table(credentials, base_id, table_name, table_fields)
@@ -73,7 +89,7 @@ async def test_invalid_field_type():
api_key=SecretStr(key),
)
postfix = uuid4().hex[:4]
base_id = "appSbaQLkcYiIOqux"
base_id = "appZPxegHEU3kDc1S"
table_name = f"test_table_{postfix}"
table_fields = [{"name": "test_field", "type": "notValid"}]
with pytest.raises(AssertionError):
@@ -91,7 +107,7 @@ async def test_create_and_update_field():
api_key=SecretStr(key),
)
postfix = uuid4().hex[:4]
base_id = "appSbaQLkcYiIOqux"
base_id = "appZPxegHEU3kDc1S"
table_name = f"test_table_{postfix}"
table_fields = [{"name": "test_field", "type": "singleLineText"}]
table = await create_table(credentials, base_id, table_name, table_fields)
@@ -133,7 +149,7 @@ async def test_record_management():
api_key=SecretStr(key),
)
postfix = uuid4().hex[:4]
base_id = "appSbaQLkcYiIOqux"
base_id = "appZPxegHEU3kDc1S"
table_name = f"test_table_{postfix}"
table_fields = [{"name": "test_field", "type": "singleLineText"}]
table = await create_table(credentials, base_id, table_name, table_fields)
@@ -261,7 +277,7 @@ async def test_webhook_management():
api_key=SecretStr(key),
)
postfix = uuid4().hex[:4]
base_id = "appSbaQLkcYiIOqux"
base_id = "appZPxegHEU3kDc1S"
table_name = f"test_table_{postfix}"
table_fields = [{"name": "test_field", "type": "singleLineText"}]
table = await create_table(credentials, base_id, table_name, table_fields)

122
autogpt_platform/backend/backend/blocks/airtable/bases.py Normal file
View File

@@ -0,0 +1,122 @@
"""
Airtable base operation blocks.
"""
from typing import Optional
from backend.sdk import (
APIKeyCredentials,
Block,
BlockCategory,
BlockOutput,
BlockSchema,
CredentialsMetaInput,
SchemaField,
)
from ._api import create_base, list_bases
from ._config import airtable
class AirtableCreateBaseBlock(Block):
"""
Creates a new base in an Airtable workspace.
"""
class Input(BlockSchema):
credentials: CredentialsMetaInput = airtable.credentials_field(
description="Airtable API credentials"
)
workspace_id: str = SchemaField(
description="The workspace ID where the base will be created"
)
name: str = SchemaField(description="The name of the new base")
tables: list[dict] = SchemaField(
description="At least one table and field must be specified. Array of table objects to create in the base. Each table should have 'name' and 'fields' properties",
default=[
{
"description": "Default table",
"name": "Default table",
"fields": [
{
"name": "ID",
"type": "number",
"description": "Auto-incrementing ID field",
"options": {"precision": 0},
}
],
}
],
)
class Output(BlockSchema):
base_id: str = SchemaField(description="The ID of the created base")
tables: list[dict] = SchemaField(description="Array of table objects")
table: dict = SchemaField(description="A single table object")
def __init__(self):
super().__init__(
id="f59b88a8-54ce-4676-a508-fd614b4e8dce",
description="Create a new base in Airtable",
categories={BlockCategory.DATA},
input_schema=self.Input,
output_schema=self.Output,
)
async def run(
self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
) -> BlockOutput:
data = await create_base(
credentials,
input_data.workspace_id,
input_data.name,
input_data.tables,
)
yield "base_id", data.get("id", None)
yield "tables", data.get("tables", [])
for table in data.get("tables", []):
yield "table", table
class AirtableListBasesBlock(Block):
"""
Lists all bases in an Airtable workspace that the user has access to.
"""
class Input(BlockSchema):
credentials: CredentialsMetaInput = airtable.credentials_field(
description="Airtable API credentials"
)
trigger: str = SchemaField(
description="Trigger the block to run - value is ignored", default="manual"
)
offset: str = SchemaField(
description="Pagination offset from previous request", default=""
)
class Output(BlockSchema):
bases: list[dict] = SchemaField(description="Array of base objects")
offset: Optional[str] = SchemaField(
description="Offset for next page (null if no more bases)", default=None
)
def __init__(self):
super().__init__(
id="4bd8d466-ed5d-4e44-8083-97f25a8044e7",
description="List all bases in Airtable",
categories={BlockCategory.DATA},
input_schema=self.Input,
output_schema=self.Output,
)
async def run(
self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
) -> BlockOutput:
data = await list_bases(
credentials,
offset=input_data.offset if input_data.offset else None,
)
yield "bases", data.get("bases", [])
yield "offset", data.get("offset", None)