docs(platform): Example block

autogpt_platform/backend/.env.example

@@ -173,6 +173,9 @@ EXA_API_KEY=
 # E2B
 E2B_API_KEY=
 
+# Example API Key
+EXAMPLE_API_KEY=
+
 # Mem0
 MEM0_API_KEY=
 

autogpt_platform/backend/backend/blocks/example/_api.py

@@ -0,0 +1,137 @@
+"""
+API module for Example API integration.
+
+This module provides a example of how to create a client for an API.
+"""
+
+# We also have a Json Wrapper library available in backend.util.json
+from json import JSONDecodeError
+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+from backend.data.model import APIKeyCredentials
+
+# This is a wrapper around the requests library that is used to make API requests.
+from backend.util.request import Requests
+
+
+class ExampleAPIException(Exception):
+    def __init__(self, message: str, status_code: int):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class CreateResourceResponse(BaseModel):
+    message: str
+    is_funny: bool
+
+
+class GetResourceResponse(BaseModel):
+    message: str
+    is_funny: bool
+
+
+class ExampleClient:
+    """Client for the Example API"""
+
+    API_BASE_URL = "https://api.example.com/v1"
+
+    def __init__(
+        self,
+        credentials: Optional[APIKeyCredentials] = None,
+        custom_requests: Optional[Requests] = None,
+    ):
+        if custom_requests:
+            self._requests = custom_requests
+        else:
+            headers: dict[str, str] = {
+                "Content-Type": "application/json",
+            }
+            if credentials:
+                headers["Authorization"] = credentials.auth_header()
+
+            self._requests = Requests(
+                extra_headers=headers,
+                raise_for_status=False,
+            )
+
+    @staticmethod
+    def _handle_response(response) -> Any:
+        """
+        Handles API response and checks for errors.
+
+        Args:
+            response: The response object from the request.
+
+        Returns:
+            The parsed JSON response data.
+
+        Raises:
+            ExampleAPIException: If the API request fails.
+        """
+        if not response.ok:
+            try:
+                error_data = response.json()
+                error_message = error_data.get("error", {}).get("message", "")
+            except JSONDecodeError:
+                error_message = response.text
+
+            raise ExampleAPIException(
+                f"Example API request failed ({response.status_code}): {error_message}",
+                response.status_code,
+            )
+
+        response_data = response.json()
+        if "errors" in response_data:
+            # This is an example error and needs to be
+            # replaced with how the real API returns errors
+            error_messages = [
+                error.get("message", "") for error in response_data["errors"]
+            ]
+            raise ExampleAPIException(
+                f"Example API returned errors: {', '.join(error_messages)}",
+                response.status_code,
+            )
+
+        return response_data
+
+    def get_resource(self, resource_id: str) -> GetResourceResponse:
+        """
+        Fetches a resource from the Example API.
+
+        Args:
+            resource_id: The ID of the resource to fetch.
+
+        Returns:
+            The resource data as a GetResourceResponse object.
+
+        Raises:
+            ExampleAPIException: If the API request fails.
+        """
+        try:
+            response = self._requests.get(
+                f"{self.API_BASE_URL}/resources/{resource_id}"
+            )
+            return GetResourceResponse(**self._handle_response(response))
+        except Exception as e:
+            raise ExampleAPIException(f"Failed to get resource: {str(e)}", 500)
+
+    def create_resource(self, data: dict) -> CreateResourceResponse:
+        """
+        Creates a new resource via the Example API.
+
+        Args:
+            data: The resource data to create.
+
+        Returns:
+            The created resource data as a CreateResourceResponse object.
+
+        Raises:
+            ExampleAPIException: If the API request fails.
+        """
+        try:
+            response = self._requests.post(f"{self.API_BASE_URL}/resources", json=data)
+            return CreateResourceResponse(**self._handle_response(response))
+        except Exception as e:
+            raise ExampleAPIException(f"Failed to create resource: {str(e)}", 500)

autogpt_platform/backend/backend/blocks/example/_auth.py

@@ -0,0 +1,37 @@
+"""
+Authentication module for Example API integration.
+
+This module provides credential types and test credentials for the Example API integration.
+It defines the structure for API key credentials used to authenticate with the Example API
+and provides mock credentials for testing purposes.
+"""
+
+from typing import Literal
+
+from pydantic import SecretStr
+
+from backend.data.model import APIKeyCredentials, CredentialsMetaInput
+from backend.integrations.providers import ProviderName
+
+# Define the type of credentials input expected for Example API
+ExampleCredentialsInput = CredentialsMetaInput[
+    Literal[ProviderName.EXAMPLE_PROVIDER], Literal["api_key"]
+]
+
+
+# Mock credentials for testing Example API integration
+TEST_CREDENTIALS = APIKeyCredentials(
+    id="9191c4f0-498f-4235-a79c-59c0e37454d4",
+    provider="example-provider",
+    api_key=SecretStr("mock-example-api-key"),
+    title="Mock Example API key",
+    expires_at=None,
+)
+
+# Dictionary representation of test credentials for input fields
+TEST_CREDENTIALS_INPUT = {
+    "provider": TEST_CREDENTIALS.provider,
+    "id": TEST_CREDENTIALS.id,
+    "type": TEST_CREDENTIALS.type,
+    "title": TEST_CREDENTIALS.title,
+}

autogpt_platform/backend/backend/blocks/example/example.py

@@ -0,0 +1,154 @@
+import logging
+
+from pydantic import BaseModel
+
+from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
+from backend.data.model import APIKeyCredentials, CredentialsField, SchemaField
+
+from ._api import ExampleClient
+from ._auth import TEST_CREDENTIALS, TEST_CREDENTIALS_INPUT, ExampleCredentialsInput
+
+logger = logging.getLogger(__name__)
+
+
+class GreetingMessage(BaseModel):
+    message: str
+    is_funny: bool
+
+
+class ExampleBlock(Block):
+
+    class Input(BlockSchema):
+        name: str = SchemaField(
+            description="The name of the example block", placeholder="Enter a name"
+        )
+        greetings: list[str] = SchemaField(
+            description="The greetings to display", default=["Hello", "Hi", "Hey"]
+        )
+        is_funny: bool = SchemaField(
+            description="Whether the block is funny",
+            placeholder="True",
+            default=True,
+            # Advanced fields are moved to the "Advanced" dropdown in the UI
+            advanced=True,
+        )
+        greeting_context: str = SchemaField(
+            description="The context of the greeting",
+            placeholder="Enter a context",
+            default="The user is looking for an inspirational greeting",
+            # Hidden fields are not shown in the UI at all
+            hidden=True,
+        )
+        # Only if the block needs credentials
+        credentials: ExampleCredentialsInput = CredentialsField(
+            description="The credentials for the example block"
+        )
+
+    class Output(BlockSchema):
+        response: GreetingMessage = SchemaField(
+            description="The response object generated by the example block."
+        )
+        all_responses: list[GreetingMessage] = SchemaField(
+            description="All the responses from the example block."
+        )
+        greeting_count: int = SchemaField(
+            description="The number of greetings in the input."
+        )
+        error: str = SchemaField(description="The error from the example block")
+
+    def __init__(self):
+        super().__init__(
+            # The unique identifier for the block, this value will be persisted in the DB.
+            # It should be unique and constant across the application run.
+            # Use the UUID format for the ID.
+            id="380694d5-3b2e-4130-bced-b43752b70de9",
+            # The description of the block, explaining what the block does.
+            description="The example block",
+            # The set of categories that the block belongs to.
+            # Each category is an instance of BlockCategory Enum.
+            categories={BlockCategory.BASIC},
+            # The schema, defined as a Pydantic model, for the input data.
+            input_schema=ExampleBlock.Input,
+            # The schema, defined as a Pydantic model, for the output data.
+            output_schema=ExampleBlock.Output,
+            # The list or single sample input data for the block, for testing.
+            # This is an instance of the Input schema with sample values.
+            test_input={
+                "name": "Craig",
+                "greetings": ["Hello", "Hi", "Hey"],
+                "is_funny": True,
+                "credentials": TEST_CREDENTIALS_INPUT,
+            },
+            # The list or single expected output if the test_input is run.
+            # Each output is a tuple of (output_name, output_data).
+            test_output=[
+                ("response", GreetingMessage(message="Hello, world!", is_funny=True)),
+                (
+                    "response",
+                    GreetingMessage(message="Hello, world!", is_funny=True),
+                ),  # We mock the function
+                (
+                    "response",
+                    GreetingMessage(message="Hello, world!", is_funny=True),
+                ),  # We mock the function
+                (
+                    "all_responses",
+                    [
+                        GreetingMessage(message="Hello, world!", is_funny=True),
+                        GreetingMessage(message="Hello, world!", is_funny=True),
+                        GreetingMessage(message="Hello, world!", is_funny=True),
+                    ],
+                ),
+                ("greeting_count", 3),
+            ],
+            # Function names on the block implementation to mock on test run.
+            # Each mock is a dictionary with function names as keys and mock implementations as values.
+            test_mock={
+                "my_function_that_can_be_mocked": lambda *args, **kwargs: GreetingMessage(
+                    message="Hello, world!", is_funny=True
+                )
+            },
+            # The credentials required for testing the block.
+            # This is an instance of APIKeyCredentials with sample values.
+            test_credentials=TEST_CREDENTIALS,
+        )
+
+    def my_function_that_can_be_mocked(
+        self, name: str, credentials: APIKeyCredentials
+    ) -> GreetingMessage:
+        logger.info("my_function_that_can_be_mocked called with input: %s", name)
+
+        # Use the ExampleClient from _api.py to make an API call
+        client = ExampleClient(credentials=credentials)
+
+        # Create a sample resource using the client
+        resource_data = {"name": name, "type": "greeting"}
+        # If your API response object matches the return type of the function,
+        # there is no need to convert the object. In this case we have a different
+        # object type for the response and the return type of the function.
+        return GreetingMessage(**client.create_resource(resource_data).model_dump())
+
+    def run(
+        self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
+    ) -> BlockOutput:
+        """
+        The run function implements the block's core logic. It processes the input_data
+        and yields the block's output.
+
+        In addition to credentials, the following parameters can be specified:
+            graph_id: The ID of the graph containing this block.
+            node_id: The ID of this block's node in the graph.
+            graph_exec_id: The ID of the current graph execution.
+            node_exec_id: The ID of the current node execution.
+            user_id: The ID of the user executing the block.
+        """
+        rtn_all_responses: list[GreetingMessage] = []
+        # Here we deomonstrate best practice for blocks that need to yield multiple items.
+        # We yield each item from the list to allow for operations on each element.
+        # We also yield the complete list for situations when the full list is needed.
+        for greeting in input_data.greetings:
+            message = self.my_function_that_can_be_mocked(greeting, credentials)
+            rtn_all_responses.append(message)
+            yield "response", message
+        yield "all_responses", rtn_all_responses
+        yield "greeting_count", len(input_data.greetings)