AutoGPT - Building Blocks

Worum geht's?

Ein Block ist die kleinste Funktionseinheit in AutoGPT — ein Block erledigt genau eine Sache. Jeder Agent ist nur eine Verschaltung von Blöcken. Diese Seite ist die praxisnahe Anleitung zum Schreiben eines eigenen Blocks in Python: die Basisklasse Block, typisierte Input-/Output-Schemas, der run()-Vertrag, wie Credentials injiziert werden und wie du einen Block testest und registrierst. Neu auf der Plattform? Beginne zuerst mit der Übersicht und bring über Self-Hosting eine lokale Instanz zum Laufen.

Quelle

Klassennamen, Signaturen und Feldlisten unten sind aus dem Repository zitiert: autogpt_platform/backend/backend/blocks/_base.py, backend/data/block.py und backend/data/model.py (dev-Branch, geprüft am 25. Juni 2026). Die Beispielblöcke sind eigens für diese Anleitung geschrieben. Gleiche Signaturen immer mit deinem Checkout ab — das SDK entwickelt sich weiter.

1. Wo Blöcke liegen

Blöcke sind Python-Module unter:

autogpt_platform/backend/backend/blocks/

Jede .py-Datei dort kann eine oder mehrere Block-Klassen definieren. Die Basisklasse und die Kerntypen stammen aus _base.py:

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

Es gibt keine manuelle Registry, die du pflegen müsstest. Blöcke werden automatisch erkannt: Beim Start sammelt initialize_blocks() jede Block-Subklasse über get_blocks() ein und synchronisiert jede davon anhand ihrer id und ihres name mit der Datenbank. Lege eine wohlgeformte Block-Klasse im Paket blocks/ an, und sie taucht im Agent Builder auf.

2. Die vier Bestandteile eines Blocks

Jeder Block ist eine Klasse, die von Block erbt und genau vier Dinge enthält:

Bestandteil	Was es ist
Input	Eine verschachtelte Klasse, die BlockSchemaInput erweitert — die typisierten Inputs, deklariert mit SchemaField.
Output	Eine verschachtelte Klasse, die BlockSchemaOutput erweitert — die typisierten Outputs (plus ein eingebautes error-Feld).
__init__	Ruft super().__init__(...) mit id, Schemas, Beschreibung, Kategorien und Testdaten des Blocks auf.
run	Ein async Generator, der die Arbeit erledigt und (output_name, value)-Paare yieldt.

Block selbst ist generisch über seine Schemas:

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

3. Ein vollständiger minimaler Block

Hier ist ein vollständiger, funktionierender Block, der für diese Anleitung geschrieben wurde. Er nimmt einen String entgegen und gibt Wort- und Zeichenanzahl zurück. Lies ihn einmal durch, danach sezieren wir jeden Bestandteil.

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",  # eine feste, eindeutige 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)

Das ist ein vollständiger Block. Beachte die vier Bestandteile, die UUID id, das set von Kategorien und dass run Ergebnisse yieldt, statt sie zurückzugeben.

4. Input- & Output-Schemas

Schemas sind Pydantic-Modelle. BlockSchemaInput und BlockSchemaOutput sind die Basisklassen, die du erweiterst; die Plattform wandelt sie in JSON Schema um, das der Agent Builder als Formular rendert und der Executor zur Validierung heranzieht.

SchemaField

Deklariere jedes Feld mit SchemaField(...) statt mit einem nackten Pydantic-Field. Seine Signatur:

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,
): ...

Die, zu denen du am häufigsten greifst:

Argument	Wirkung
description	Wird als Hilfetext im Builder angezeigt. Setze sie immer — sie ist das, was Nutzer (und Agenten bauende LLMs) lesen.
default / default_factory	Ein Standardwert. Ein Feld ohne Default ist erforderlich.
placeholder	Beispieltext im Eingabe-Widget.
advanced	Versteckt das Feld hinter „Erweitert" in der UI. Automatisch gesetzt: Felder mit Default stehen standardmäßig auf advanced=True, erforderliche Felder auf advanced=False.
secret	Markiert den Wert als sensibel (maskiert, nicht protokolliert).
ge / le	Numerische Schranken (≥ / ≤). Werden zur Validierungszeit erzwungen.
min_length / max_length	Längenschranken für Strings/Collections.
hidden	Hält das Feld vollständig aus der UI heraus.

Erforderlich vs. optional

Ein Feld ist erforderlich, wenn es weder default noch default_factory hat. Gib einem Feld einen Default, um es optional zu machen. Dieselbe Regel steuert den advanced-Auto-Default, sodass erforderliche Felder sichtbar bleiben und optionale sich wegklappen.

Der eingebaute error-Output

BlockSchemaOutput deklariert bereits ein Feld für dich:

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

So kann jeder Block einen error-Output ausgeben, ohne ihn zu deklarieren. Dieser Output ist besonders — siehe §6.

5. __init__ — Metadaten registrieren

__init__ nimmt keine Argumente vom Aufrufer entgegen; es ruft super().__init__(...) mit den statischen Metadaten des Blocks auf. Die Signatur des Konstruktors:

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	Hinweise
id	Eine eindeutige, konstante UUID. Sie wird in der DB persistiert und darf sich für einen gegebenen Block niemals ändern, sonst gehen bestehende Agenten kaputt. Generiere sie einmal (z. B. python -c "import uuid; print(uuid.uuid4())") und schreibe sie fest.
description	Ein Satz dazu, was der Block tut. Erscheint im Builder und bei Agenten bauenden LLMs.
categories	Ein set von BlockCategory (siehe §7). Steuert Gruppierung/Suche im Builder.
input_schema / output_schema	Deine verschachtelten Input- / Output-Klassen.
test_input / test_output / test_mock / test_credentials	Test-Fixtures (siehe §8).
disabled	Bei True wird der Block von der Ausführung ausgeblendet — praktisch, um einen Stub auszuliefern oder per Konfiguration zu sperren.
static_output	Ob die Output-Links des Blocks standardmäßig statisch sind.
block_type	Üblicherweise auf STANDARD belassen; spezielle Typen existieren für Input-/Output-/Webhook-/AI-Blöcke.
is_sensitive_action	Markiert den Block für eine optionale Human-in-the-Loop-Prüfung, bevor er ausgeführt wird.

6. Der run()-Vertrag

In run passiert die eigentliche Arbeit. Die Regeln:

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

1. Es ist async und ein Generator. Du yieldst; du gibst keinen Wert per return zurück. BlockOutput ist AsyncGenerator[tuple[str, Any], None].
2. Du yieldest (name, value)-Tupel. name muss zu einem in deinem Output-Schema deklarierten Feld passen. Der Wert wird bei STANDARD-Blöcken gegen den Typ dieses Feldes validiert.
3. Du kannst mehrfach yielden. Ein Pin kann über die Zeit mehrere Werte ausgeben, und du kannst an verschiedene Output-Pins yielden. Das ist das Streaming-Modell, das Blöcke komponierbar macht.
4. input_data ist deine typisierte Input-Instanz — greife auf Felder als Attribute zu (input_data.text), bereits validiert und mit angewendeten Defaults.
5. kwargs trägt den Ausführungskontext. Der Executor übergibt graph_id, node_id, graph_exec_id, node_exec_id und user_id, plus alle aufgelösten Credentials (siehe §9). Akzeptiere **kwargs, auch wenn du sie ignorierst.

Fehlerbehandlung — zwei Muster

Muster A — "error" yielden. Das Yielden des speziellen error-Outputs lässt den Executor einen BlockExecutionError auslösen und den Node stoppen:

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

Muster B — Exceptions durchreichen lassen. Jede Exception, die du auslöst (oder die hochblubbert), wird vom Framework in einen BlockExecutionError / BlockUnknownError verpackt, mit Name und id deines Blocks angehängt. Du musst also nicht alles abfangen — einen klaren ValueError auszulösen ist in Ordnung.

error ist reserviert

Verwende error nicht als Namen eines normalen Daten-Outputs. Das Yielden von ("error", …) bricht den Node immer ab. Wenn dein Block einen erwarteten „Fehlschlag"-Zweig hat, auf den nachgelagerte Nodes reagieren sollen, modelliere ihn als eigenen benannten Output (z. B. not_found), nicht als error.

7. Kategorien & Block-Typen

BlockCategory

Wähle eine oder mehrere aus dem Enum (jeder Wert ist zugleich seine menschenlesbare Beschreibung):

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

Die meisten Blöcke sind STANDARD. Die Laufzeit definiert außerdem: INPUT, OUTPUT, NOTE, WEBHOOK, WEBHOOK_MANUAL, AGENT, AI, AYRSHARE, HUMAN_IN_THE_LOOP, MCP_TOOL. Du setzt das selten direkt — zum Beispiel kippt die Angabe einer webhook_config den Block automatisch auf einen Webhook-Typ.

8. Einen Block testen

Blöcke liefern ihre eigenen Test-Fixtures in __init__ mit, sodass die Block-Testsuite der Plattform sie automatisch durchlaufen kann. Es gibt vier Felder.

test_input und test_output

test_input ist das Input-Dict (oder eine Liste von Dicts). test_output ist das erwartete (name, value)-Paar (bzw. die Paare). Der Test führt run() mit dem Input aus und prüft, ob die Yields übereinstimmen.

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

test_output-Einträge können statt eines Literals auch ein Callable verwenden, für Werte, die du nicht exakt festnageln kannst (Zeitstempel, generierter Text):

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

test_mock

Für Blöcke, die das Netzwerk oder anderen seiteneffektbehafteten Code aufrufen, ersetzt test_mock benannte Methoden auf deinem Block während des Tests, sodass Tests deterministisch und offline bleiben. Es ordnet einem Methodennamen seinen Ersatz zu:

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:
    ...  # echter HTTP-Aufruf, im Test durch den Mock ersetzt

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

test_credentials

Für Blöcke mit einem Credentials-Feld (§9) gibst du hier ein gefälschtes Credentials-Objekt an, damit der Test ohne echte Secrets laufen kann.

9. Credentials & Authentifizierung

Blöcke, die authentifizierte APIs aufrufen, deklarieren ein Credentials-Feld. Das Framework erzwingt eine strikte Benennungs-/Typisierungsregel auf BlockSchema-Subklassen:

Ein als CredentialsMetaInput annotiertes Feld muss credentials oder *_credentials heißen, und jedes Feld namens credentials / *_credentials muss vom Typ CredentialsMetaInput sein.

Deklariere es mit 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()
        # … rufe die API mit api_key auf …
        yield "result", "…"

Wie es zur Laufzeit funktioniert:

• Der Nutzer wählt/gibt Credentials für dieses Feld im Builder ein; die Plattform speichert sie verschlüsselt.
• Bei der Ausführung löst das Framework die Credential auf und injiziert das tatsächliche Credentials-Objekt als Keyword-Argument passend zum Feldnamen (credentials oben) — deshalb akzeptiert run credentials: APIKeyCredentials.
• Secret-Werte sind in SecretStr verpackt; lies sie mit .get_secret_value().

Verfügbare Credential-Objekttypen aus backend.data.model:

Typ	Wofür
APIKeyCredentials	Einfache API-Key-Authentifizierung (.api_key).
OAuth2Credentials	OAuth2-Flows (.access_token, .refresh_token, .scopes, .auth_header()).
UserPasswordCredentials	Benutzername-/Passwort-Authentifizierung.

Lies Secrets niemals in Logs oder Outputs

Hole ein Secret mit .get_secret_value() nur an der Stelle heraus, an der du es für den API-Aufruf brauchst. yielde es niemals, protokolliere es nicht und packe es nicht in eine Fehlermeldung. Die Plattform maskiert SecretStr überall sonst mit Absicht.

10. Kosten (optional)

Wenn dein Block kostenpflichtige Ressourcen verbraucht, kann er einen Credit-Kostensatz tragen, sodass Self-Hoster und die Cloud ihn abrechnen können. Kosten werden mit BlockCost / BlockCostType ausgedrückt, wobei der Typ RUN (pro Ausführung), BYTE, SECOND, ITEMS, COST_USD oder TOKENS (modellbezogene LLM-Tarifs­tabellen) sein kann. Das ist ein fortgeschrittenes Thema — die meisten einfachen Blöcke brauchen gar keine Kosten. Details siehe _base.py (BlockCost, BlockCostType) und backend/data/credit.py.

11. Checkliste & Best Practices

Bevor du einen PR mit einem neuen Block öffnest:

• Eine Verantwortlichkeit. Der Block tut eine Sache. Wenn er drei tut, mach drei Blöcke.
• Stabile UUID id. Einmal generiert, fest eingetragen, niemals geändert.
• Jedes Feld hat eine description. Nutzer und Agenten bauende LLMs hängen davon ab.
• Sinnvolle Defaults. Erforderliche Felder haben keine; optionale schon.
• run yieldet, gibt nicht zurück. Und es akzeptiert **kwargs.
• Output-Namen passen zum Output-Schema. Reserviere error für Fehler.
• Secrets über Credentials-Felder namens credentials / *_credentials, gelesen mit .get_secret_value().
• Tests bereitgestellt — test_input + test_output, plus test_mock für jeden Netzwerk-/seiteneffektbehafteten Aufruf und test_credentials, falls der Block sich authentifiziert.
• Kategorisiert mit dem passenden BlockCategory-Set.

12. Quellen

• AutoGPT-Anleitung „new blocks" — 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, geprüft 2026-06-25)