AutoGPT - Building Blocks

What's this about?

A block is the smallest unit of functionality in AutoGPT — one block does one thing. Every agent is just blocks wired together. This page is the hands-on guide to writing your own block in Python: the Block base class, typed input/output schemas, the run() contract, how credentials are injected, and how to test and register a block. New to the platform? Start with the overview and get a local instance running via Self-Hosting first.

Source

Class names, signatures and field lists below are quoted from the repository: autogpt_platform/backend/backend/blocks/_base.py, backend/data/block.py and backend/data/model.py (dev branch, reviewed June 25, 2026). The example blocks are original and written for this guide. Always cross-check signatures against your checkout — the SDK evolves.

1. Where blocks live

Blocks are Python modules under:

autogpt_platform/backend/backend/blocks/

Each .py file there can define one or more block classes. The base class and the core types come from _base.py:

from backend.blocks._base import (
    Block,
    BlockCategory,
    BlockOutput,
    BlockSchemaInput,
    BlockSchemaOutput,
)
from backend.data.model import SchemaField

There is no manual registry to edit. Blocks are discovered automatically: at startup initialize_blocks() collects every Block subclass via get_blocks() and syncs each one to the database by its id and name. Create a well-formed block class in the blocks/ package and it shows up in the Agent Builder.

2. The four parts of a block

Every block is a class that subclasses Block and contains exactly four things:

Part	What it is
Input	A nested class extending BlockSchemaInput — the typed inputs, declared with SchemaField.
Output	A nested class extending BlockSchemaOutput — the typed outputs (plus a built-in error field).
__init__	Calls super().__init__(...) with the block's id, schemas, description, categories and test data.
run	An async generator that does the work and yields (output_name, value) pairs.

Block itself is generic over its schemas:

class Block(ABC, Generic[BlockSchemaInputType, BlockSchemaOutputType]):
    ...
    @abstractmethod
    async def run(self, input_data: BlockSchemaInputType, **kwargs) -> BlockOutput:
        ...

3. A complete minimal block

Here is a full, working block written for this guide. It takes a string and returns word and character counts. Read it once, then we'll dissect each part.

from enum import Enum

from backend.blocks._base import (
    Block,
    BlockCategory,
    BlockOutput,
    BlockSchemaInput,
    BlockSchemaOutput,
)
from backend.data.model import SchemaField


class CountMode(Enum):
    INCLUDE_WHITESPACE = "Include whitespace"
    EXCLUDE_WHITESPACE = "Exclude whitespace"


class TextStatisticsBlock(Block):
    class Input(BlockSchemaInput):
        text: str = SchemaField(
            description="The text to analyse.",
            placeholder="Paste some text…",
        )
        mode: CountMode = SchemaField(
            description="How to count characters.",
            default=CountMode.INCLUDE_WHITESPACE,
        )

    class Output(BlockSchemaOutput):
        word_count: int = SchemaField(description="Number of words.")
        char_count: int = SchemaField(description="Number of characters.")

    def __init__(self):
        super().__init__(
            id="2f1c0a7e-5b4d-4e2a-9c8f-1d3e6a7b9c01",  # a fixed, unique UUID
            description="Counts words and characters in a piece of text.",
            categories={BlockCategory.TEXT},
            input_schema=TextStatisticsBlock.Input,
            output_schema=TextStatisticsBlock.Output,
            test_input={"text": "hello brave new world"},
            test_output=[
                ("word_count", 4),
                ("char_count", 21),
            ],
        )

    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
        text = input_data.text
        yield "word_count", len(text.split())

        if input_data.mode == CountMode.EXCLUDE_WHITESPACE:
            text = "".join(text.split())
        yield "char_count", len(text)

That's a complete block. Note the four parts, the UUID id, the set of categories, and that run yields results rather than returning them.

4. Input & output schemas

Schemas are Pydantic models. BlockSchemaInput and BlockSchemaOutput are the base classes you extend; the platform turns them into JSON Schema, which the Agent Builder renders as a form and the executor validates against.

SchemaField

Declare every field with SchemaField(...) instead of a bare Pydantic Field. Its signature:

def SchemaField(
    default=PydanticUndefined,
    *,
    default_factory=None,
    title=None,
    description=None,
    placeholder=None,
    advanced=None,
    secret=False,
    exclude=False,
    hidden=None,
    depends_on=None,
    ge=None,
    le=None,
    min_length=None,
    max_length=None,
    discriminator=None,
    format=None,
    json_schema_extra=None,
): ...

The ones you'll reach for most:

Argument	Effect
description	Shown as help text in the builder. Always set it — it's what users (and LLMs building agents) read.
default / default_factory	A default value. A field with no default is required.
placeholder	Example text in the input widget.
advanced	Hides the field behind "Advanced" in the UI. Auto-set: fields with a default default to advanced=True, required fields to advanced=False.
secret	Marks the value sensitive (masked, not logged).
ge / le	Numeric bounds (≥ / ≤). Enforced at validation time.
min_length / max_length	String/collection length bounds.
hidden	Keeps the field out of the UI entirely.

Required vs optional

A field is required when it has no default and no default_factory. Give a field a default to make it optional. The same rule drives the advanced auto-default, so required fields stay visible and optional ones tuck away.

The built-in error output

BlockSchemaOutput already declares one field for you:

class BlockSchemaOutput(BlockSchema):
    error: str = SchemaField(
        description="Error message if the operation failed", default=""
    )

So every block can emit an error output without declaring it. This output is special — see §6.

5. __init__ — registering metadata

__init__ takes no arguments from the caller; it calls super().__init__(...) with the block's static metadata. The constructor signature:

def __init__(
    self,
    id: str = "",
    description: str = "",
    contributors: list[ContributorDetails] = [],
    categories: set[BlockCategory] | None = None,
    input_schema: Type[BlockSchemaInputType] = EmptyInputSchema,
    output_schema: Type[BlockSchemaOutputType] = EmptyOutputSchema,
    test_input: BlockInput | list[BlockInput] | None = None,
    test_output: BlockTestOutput | list[BlockTestOutput] | None = None,
    test_mock: dict[str, Any] | None = None,
    test_credentials: Credentials | dict[str, Credentials] | None = None,
    disabled: bool = False,
    static_output: bool = False,
    block_type: BlockType = BlockType.STANDARD,
    webhook_config: BlockWebhookConfig | BlockManualWebhookConfig | None = None,
    is_sensitive_action: bool = False,
):

Parameter	Notes
id	A unique, constant UUID. It's persisted in the DB and must never change for a given block, or existing agents break. Generate one once (e.g. python -c "import uuid; print(uuid.uuid4())") and hard-code it.
description	One sentence on what the block does. Surfaced in the builder and to agent-building LLMs.
categories	A set of BlockCategory (see §7). Drives grouping/search in the builder.
input_schema / output_schema	Your nested Input / Output classes.
test_input / test_output / test_mock / test_credentials	Test fixtures (see §8).
disabled	If True, the block is hidden from execution — handy to ship a stub or gate behind config.
static_output	Whether the block's output links are static by default.
block_type	Usually leave as STANDARD; special types exist for input/output/webhook/AI blocks.
is_sensitive_action	Marks the block for optional human-in-the-loop review before it runs.

6. The run() contract

run is where the work happens. The rules:

async def run(self, input_data: Input, **kwargs) -> BlockOutput:
    ...
    yield "output_name", value

1. It is async and a generator. You yield; you don't return a value. BlockOutput is AsyncGenerator[tuple[str, Any], None].
2. You yield (name, value) tuples. name must match a field declared in your Output schema. The value is validated against that field's type for STANDARD blocks.
3. You can yield many times. One pin can emit several values over time, and you can yield to different output pins. This is the streaming model that makes blocks composable.
4. input_data is your typed Input instance — access fields as attributes (input_data.text), already validated and with defaults applied.
5. kwargs carries execution context. The executor passes graph_id, node_id, graph_exec_id, node_exec_id and user_id, plus any resolved credentials (see §9). Accept **kwargs even if you ignore them.

Error handling — two patterns

Pattern A — yield "error". Yielding the special error output makes the executor raise a BlockExecutionError and stop the node:

async def run(self, input_data: Input, **kwargs) -> BlockOutput:
    if input_data.b == 0:
        yield "error", "Cannot divide by zero"
        return
    yield "result", input_data.a / input_data.b

Pattern B — let exceptions propagate. Any exception you raise (or that bubbles up) is wrapped by the framework into a BlockExecutionError / BlockUnknownError with your block's name and id attached. So you don't have to catch everything — raising a clear ValueError is fine.

error is reserved

Don't use error as the name of a normal data output. Yielding ("error", …) always aborts the node. If your block has an expected "failure" branch that downstream nodes should react to, model it as its own named output (e.g. not_found), not as error.

7. Categories & block types

BlockCategory

Pick one or more from the enum (each value is also its human description):

AI, SOCIAL, TEXT, SEARCH, BASIC, INPUT, OUTPUT, LOGIC, COMMUNICATION, DEVELOPER_TOOLS, DATA, HARDWARE, AGENT, CRM, SAFETY, PRODUCTIVITY, ISSUE_TRACKING, MULTIMEDIA, MARKETING.

categories={BlockCategory.TEXT, BlockCategory.DATA}

BlockType

Most blocks are STANDARD. The runtime also defines: INPUT, OUTPUT, NOTE, WEBHOOK, WEBHOOK_MANUAL, AGENT, AI, AYRSHARE, HUMAN_IN_THE_LOOP, MCP_TOOL. You rarely set this directly — for example, providing a webhook_config automatically flips the block to a webhook type.

8. Testing a block

Blocks ship their own test fixtures in __init__, so the platform's block test suite can exercise them automatically. There are four fields.

test_input and test_output

test_input is the input dict (or a list of dicts). test_output is the expected (name, value) pair(s). The test runs run() with the input and asserts the yields match.

test_input={"text": "hello brave new world"},
test_output=[
    ("word_count", 4),
    ("char_count", 21),
],

test_output entries can also use a callable instead of a literal, for values you can't pin down exactly (timestamps, generated text):

test_output=[
    ("word_count", lambda v: isinstance(v, int) and v > 0),
]

test_mock

For blocks that call the network or other side-effecting code, test_mock replaces named methods on your block during the test, so tests stay deterministic and offline. It maps a method name to its replacement:

class Input(BlockSchemaInput):
    city: str = SchemaField(description="City to look up.")

class Output(BlockSchemaOutput):
    temperature: float = SchemaField(description="Current temperature in °C.")

def __init__(self):
    super().__init__(
        id="…",
        input_schema=WeatherBlock.Input,
        output_schema=WeatherBlock.Output,
        categories={BlockCategory.SEARCH},
        test_input={"city": "Münster"},
        test_output=[("temperature", 17.5)],
        test_mock={"_fetch_temperature": lambda city: 17.5},
    )

async def _fetch_temperature(self, city: str) -> float:
    ...  # real HTTP call, replaced by the mock during tests

async def run(self, input_data: Input, **kwargs) -> BlockOutput:
    yield "temperature", await self._fetch_temperature(input_data.city)

test_credentials

For blocks with a credentials field (§9), supply a fake Credentials object here so the test can run without real secrets.

9. Credentials & authentication

Blocks that call authenticated APIs declare a credentials field. The framework enforces a strict naming/typing rule on BlockSchema subclasses:

A field annotated as CredentialsMetaInput must be named credentials or *_credentials, and any field named credentials / *_credentials must be of type CredentialsMetaInput.

Declare it with CredentialsField:

from backend.data.model import (
    APIKeyCredentials,
    CredentialsField,
    CredentialsMetaInput,
    SchemaField,
)


class MyApiBlock(Block):
    class Input(BlockSchemaInput):
        credentials: CredentialsMetaInput = CredentialsField(
            description="API key for the service.",
        )
        query: str = SchemaField(description="What to search for.")

    class Output(BlockSchemaOutput):
        result: str = SchemaField(description="The API response.")

    async def run(
        self,
        input_data: Input,
        *,
        credentials: APIKeyCredentials,
        **kwargs,
    ) -> BlockOutput:
        api_key = credentials.api_key.get_secret_value()
        # … call the API with api_key …
        yield "result", "…"

How it works at run time:

• The user selects/enters credentials for that field in the builder; the platform stores them encrypted.
• At execution the framework resolves the credential and injects the actual Credentials object as a keyword argument matching the field name (credentials above) — that's why run accepts credentials: APIKeyCredentials.
• Secret values are wrapped in SecretStr; read them with .get_secret_value().

Credential object types available from backend.data.model:

Type	For
APIKeyCredentials	Simple API-key auth (.api_key).
OAuth2Credentials	OAuth2 flows (.access_token, .refresh_token, .scopes, .auth_header()).
UserPasswordCredentials	Username/password auth.

Never read secrets into logs or outputs

Pull a secret out with .get_secret_value() only at the point you need it for the API call. Never yield it, log it, or put it in an error message. The platform masks SecretStr everywhere else on purpose.

10. Costs (optional)

If your block consumes paid resources, it can carry a credit cost so self-hosters and the cloud can meter it. Costs are expressed with BlockCost / BlockCostType, where the type can be RUN (per execution), BYTE, SECOND, ITEMS, COST_USD or TOKENS (per-model LLM rate tables). This is an advanced topic — most simple blocks need no cost at all. See _base.py (BlockCost, BlockCostType) and backend/data/credit.py for the details.

11. Checklist & best practices

Before you open a PR with a new block:

• One responsibility. The block does one thing. If it does three, make three blocks.
• Stable UUID id. Generated once, hard-coded, never changed.
• Every field has a description. Users and agent-building LLMs depend on it.
• Sensible defaults. Required fields have none; optional fields do.
• run yields, doesn't return. And it accepts **kwargs.
• Output names match the Output schema. Reserve error for failures.
• Secrets via credentials fields named credentials / *_credentials, read with .get_secret_value().
• Tests provided — test_input + test_output, plus test_mock for any network/side-effecting call and test_credentials if the block authenticates.
• Categorised with the right BlockCategory set.

12. Sources

• AutoGPT "new blocks" guide — agpt.co/docs/platform/building-blocks/new_blocks
• Significant-Gravitas/AutoGPT — autogpt_platform/backend/backend/blocks/_base.py, backend/data/block.py, backend/data/model.py, backend/blocks/maths.py (dev branch, reviewed 2026-06-25)