Block Creation with SDK

This guide explains how to create new blocks for the AutoGPT Platform using the SDK pattern with advanced features.

Overview

Blocks are reusable components that perform specific tasks in AutoGPT workflows. They can integrate with external services, process data, or perform any programmatic operation.

Basic Structure

1. Create Provider Configuration

First, create a _config.py file to configure your provider using the ProviderBuilder:

python

from backend.sdk import BlockCostType, ProviderBuilder

# Simple API key provider
my_provider = (
    ProviderBuilder("my_provider")
    .with_api_key("MY_PROVIDER_API_KEY", "My Provider API Key")
    .with_base_cost(1, BlockCostType.RUN)
    .build()
)

For OAuth providers:

python

from backend.sdk import BlockCostType, ProviderBuilder
from ._oauth import MyProviderOAuthHandler

my_provider = (
    ProviderBuilder("my_provider")
    .with_oauth(
        MyProviderOAuthHandler,
        scopes=["read", "write"],
        client_id_env_var="MY_PROVIDER_CLIENT_ID",
        client_secret_env_var="MY_PROVIDER_CLIENT_SECRET",
    )
    .with_base_cost(1, BlockCostType.RUN)
    .build()
)

2. Create the Block Class

Create your block file (e.g., my_block.py):

python

import uuid
from backend.sdk import (
    APIKeyCredentials,
    Block,
    BlockCategory,
    BlockOutput,
    BlockSchema,
    BlockSchemaInput,
    BlockSchemaOutput,
    CredentialsMetaInput,
    SchemaField,
)
from ._config import my_provider


class MyBlock(Block):
    class Input(BlockSchemaInput):
        # Define input fields
        credentials: CredentialsMetaInput = my_provider.credentials_field(
            description="API credentials for My Provider"
        )
        query: str = SchemaField(description="The query to process")
        limit: int = SchemaField(
            description="Number of results", 
            default=10,
            ge=1,  # Greater than or equal to 1
            le=100  # Less than or equal to 100
        )
        advanced_option: str = SchemaField(
            description="Advanced setting",
            default="",
            advanced=True  # Hidden by default in UI
        )

    class Output(BlockSchemaOutput):
        # Define output fields
        results: list = SchemaField(description="List of results")
        count: int = SchemaField(description="Total count")
        # error output pin is already defined on BlockSchemaOutput

    def __init__(self):
        super().__init__(
            id=str(uuid.uuid4()),  # Generate unique ID
            description="Brief description of what this block does",
            categories={BlockCategory.SEARCH},  # Choose appropriate categories
            input_schema=self.Input,
            output_schema=self.Output,
        )

    async def run(
        self, 
        input_data: Input, 
        *, 
        credentials: APIKeyCredentials,
        **kwargs
    ) -> BlockOutput:
        try:
            # Your block logic here
            results = await self.process_data(
                input_data.query,
                input_data.limit,
                credentials
            )
            
            # Yield outputs
            yield "results", results
            yield "count", len(results)
            
        except Exception as e:
            yield "error", str(e)

    async def process_data(self, query, limit, credentials):
        # Implement your logic
        # Use credentials.api_key.get_secret_value() to access the API key
        pass

Key Components Explained

Provider Configuration

The ProviderBuilder allows you to:

.with_api_key(): Add API key authentication
.with_oauth(): Add OAuth authentication
.with_base_cost(): Set resource costs for the block
.with_webhook_manager(): Add webhook support
.with_user_password(): Add username/password auth

Block Schema

Input/Output classes: Define the data structure using BlockSchema
SchemaField: Define individual fields with validation
CredentialsMetaInput: Special field for handling credentials

Block Implementation

Unique ID: Generate using uuid.uuid4()
Categories: Choose from BlockCategory enum (e.g., SEARCH, AI, PRODUCTIVITY)
async run(): Main execution method that yields outputs
Error handling: Error output pin is already defined on BlockSchemaOutput

Advanced Features

Testing

Add test configuration to your block:

python

def __init__(self):
    super().__init__(
        # ... other config ...
        test_input={
            "query": "test query",
            "limit": 5,
            "credentials": {
                "provider": "my_provider",
                "id": str(uuid.uuid4()),
                "type": "api_key"
            }
        },
        test_output=[
            ("results", ["result1", "result2"]),
            ("count", 2)
        ],
        test_mock={
            "process_data": lambda *args, **kwargs: ["result1", "result2"]
        }
    )

OAuth Support

Create an OAuth handler in _oauth.py:

python

from backend.integrations.oauth.base import BaseOAuthHandler

class MyProviderOAuthHandler(BaseOAuthHandler):
    PROVIDER_NAME = "my_provider"
    
    def _get_authorization_url(self, scopes: list[str], state: str) -> str:
        # Implementation
        pass
    
    def _exchange_code_for_token(self, code: str, scopes: list[str]) -> dict:
        # Implementation
        pass

Webhook Support

Create a webhook manager in _webhook.py:

python

from backend.integrations.webhooks._base import BaseWebhooksManager

class MyProviderWebhookManager(BaseWebhooksManager):
    PROVIDER_NAME = "my_provider"
    
    async def validate_event(self, event: dict) -> bool:
        # Implementation
        pass

File Organization

backend/blocks/my_provider/
├── __init__.py          # Export your blocks
├── _config.py           # Provider configuration  
├── _oauth.py           # OAuth handler (optional)
├── _webhook.py         # Webhook manager (optional)
├── _api.py             # API client wrapper (optional)
├── models.py           # Data models (optional)
└── my_block.py         # Block implementations

Best Practices

Error Handling: Use BlockInputError for validation failures and BlockExecutionError for runtime errors (import from backend.util.exceptions). These inherit from ValueError so the executor treats them as user-fixable. See Error Handling in new_blocks.md for details.
Credentials: Use the provider's credentials_field() method
Validation: Use SchemaField constraints (ge, le, min_length, etc.)
Categories: Choose appropriate categories for discoverability
Advanced Fields: Mark complex options as advanced=True
Async Operations: Use async/await for I/O operations
API Clients: Use Requests() from SDK or external libraries
Testing: Include test inputs/outputs for validation

Common Patterns

Making API Requests

python

from backend.sdk import Requests

async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs):
    headers = {
        "Authorization": f"Bearer {credentials.api_key.get_secret_value()}",
        "Content-Type": "application/json"
    }
    
    response = await Requests().post(
        "https://api.example.com/endpoint",
        headers=headers,
        json={"query": input_data.query}
    )
    
    data = response.json()
    yield "results", data.get("results", [])

Multiple Auth Types

python

async def run(
    self, 
    input_data: Input, 
    *, 
    credentials: OAuth2Credentials | APIKeyCredentials,
    **kwargs
):
    if isinstance(credentials, OAuth2Credentials):
        # Handle OAuth
        token = credentials.access_token.get_secret_value()
    else:
        # Handle API key
        token = credentials.api_key.get_secret_value()

Handling Files

When your block works with files (images, videos, documents), use store_media_file():

python

from backend.data.execution import ExecutionContext
from backend.util.file import store_media_file
from backend.util.type import MediaFileType

async def run(
    self,
    input_data: Input,
    *,
    execution_context: ExecutionContext,
    **kwargs,
):
    # PROCESSING: Need local file path for tools like ffmpeg, MoviePy, PIL
    local_path = await store_media_file(
        file=input_data.video,
        execution_context=execution_context,
        return_format="for_local_processing",
    )

    # EXTERNAL API: Need base64 content for APIs like Replicate, OpenAI
    image_b64 = await store_media_file(
        file=input_data.image,
        execution_context=execution_context,
        return_format="for_external_api",
    )

    # OUTPUT: Return to user/next block (auto-adapts to context)
    result = await store_media_file(
        file=generated_url,
        execution_context=execution_context,
        return_format="for_block_output",  # workspace:// in CoPilot, data URI in graphs
    )
    yield "image_url", result

Return format options:

"for_local_processing" - Local file path for processing tools
"for_external_api" - Data URI for external APIs needing base64
"for_block_output" - Always use for outputs - automatically picks best format

Testing Your Block

bash

# Run all block tests
poetry run pytest backend/blocks/test/test_block.py -xvs

# Test specific block
poetry run pytest 'backend/blocks/test/test_block.py::test_available_blocks[MyBlock]' -xvs

Integration Checklist

Example Blocks for Reference

Simple API: /backend/blocks/firecrawl/ - Basic API key authentication
OAuth + API: /backend/blocks/linear/ - OAuth and API key support
Webhooks: /backend/blocks/exa/ - Includes webhook manager

Study these examples to understand different patterns and approaches for building blocks.