> For the complete documentation index, see [llms.txt](https://agpt.co/docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://agpt.co/docs/platform/building-blocks/block-sdk-guide.md).

# Block SDK Guide

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`:

!!! note "Simple API key provider" \`\`\`python from backend.sdk import BlockCostType, ProviderBuilder

````
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:

!!! note "OAuth provider configuration" \`\`\`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):
        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,
            le=100,
        )
        advanced_option: str = SchemaField(
            description="Advanced setting",
            default="",
            advanced=True,
        )

    class Output(BlockSchemaOutput):
        results: list = SchemaField(description="List of results")
        count: int = SchemaField(description="Total count")

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

    async def run(
        self,
        input_data: Input,
        *,
        credentials: APIKeyCredentials,
        **kwargs
    ) -> BlockOutput:
        try:
            results = await self.process_data(
                input_data.query,
                input_data.limit,
                credentials
            )

            yield "results", results
            yield "count", len(results)

        except Exception as e:
            yield "error", str(e)

    async def process_data(self, query, limit, credentials):
        pass
```

!!! note "Input Schema Fields" - **`credentials`**: Use `my_provider.credentials_field()` to add provider authentication - **`query`**: Simple string field with description - **`limit`**: Integer field with validation constraints (`ge=1`, `le=100`) - **`advanced_option`**: Marked with `advanced=True` to hide from basic UI

!!! note "Output Schema Fields" - **`results`**: List of results from the block - **`count`**: Total count of results - The `error` output pin is already defined on `BlockSchemaOutput`

!!! note "Block Initialization" - **`id`**: Generate a unique ID using `uuid.uuid4()` - **`description`**: Brief description of what the block does - **`categories`**: Choose from `BlockCategory` enum (e.g., SEARCH, AI, PRODUCTIVITY) - **`input_schema` / `output_schema`**: Assign the Input and Output classes

!!! note "Run Method" - Implement your block logic in `process_data()` helper method - Use `credentials.api_key.get_secret_value()` to access the API key - Use `yield` to output results

## 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

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

## Advanced Features

### Testing

Add test configuration to your block:

!!! note "Test Configuration" `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`:

!!! note "OAuth Handler Implementation" \`\`\`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:
        # Implement URL generation for OAuth flow
        pass

    def _exchange_code_for_token(self, code: str, scopes: list[str]) -> dict:
        # Implement token exchange logic
        pass
```
````

### Webhook Support

Create a webhook manager in `_webhook.py`:

!!! note "Webhook Manager Implementation" \`\`\`python from backend.integrations.webhooks.\_base import BaseWebhooksManager

````
class MyProviderWebhookManager(BaseWebhooksManager):
    PROVIDER_NAME = "my_provider"

    async def validate_event(self, event: dict) -> bool:
        # Implement event validation logic
        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

1. **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](/docs/platform/building-blocks/new_blocks.md#error-handling) for details.
2. **Credentials**: Use the provider's `credentials_field()` method
3. **Validation**: Use SchemaField constraints (ge, le, min\_length, etc.)
4. **Categories**: Choose appropriate categories for discoverability
5. **Advanced Fields**: Mark complex options as `advanced=True`
6. **Async Operations**: Use `async`/`await` for I/O operations
7. **API Clients**: Use `Requests()` from SDK or external libraries
8. **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 credentials
        token = credentials.access_token.get_secret_value()
    else:
        # Handle API key credentials
        token = credentials.api_key.get_secret_value()
```

!!! note "Authentication Types" - **`OAuth2Credentials`**: Access token via `credentials.access_token.get_secret_value()` - **`APIKeyCredentials`**: API key via `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",
    )
    yield "image_url", result
```

!!! note "File Handling Patterns" - **PROCESSING**: Use `"for_local_processing"` when you need a local file path for tools like ffmpeg, MoviePy, PIL - **EXTERNAL API**: Use `"for_external_api"` when sending content to APIs like Replicate or OpenAI (returns base64 data URI) - **OUTPUT**: Use `"for_block_output"` to return results - this automatically adapts: `workspace://` in CoPilot, data URI in graphs

**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

* [ ] Create provider configuration in `_config.py`
* [ ] Implement block class with Input/Output schemas
* [ ] Generate unique block ID with `uuid.uuid4()`
* [ ] Choose appropriate block categories
* [ ] Implement `async run()` method
* [ ] Handle errors gracefully
* [ ] Add test configuration
* [ ] Export block in `__init__.py`
* [ ] Test the block
* [ ] Document any special requirements

## 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.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://agpt.co/docs/platform/building-blocks/block-sdk-guide.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.