A User-Defined Function (UDF) is a custom function that you can use as an action in your workflows. All Tracecat Actions use this underlying UDF framework to perform tasks.

Motivations

  • Automatically generate input/output validators for frontend actions through backend code.
  • Provide a customizable and extensible framework for defining actions.

Framework

We’ve developed a simple framework that takes UDFs and converts them into Temporal activities and UI action blocks. The framework currently only supports using Python functions as UDFs.

Registry

The registry singleton is a global Python object that stores all the decorated UDFs. At runtime, Tracecat’s workflow runner dynamically executes these functions according to the workflow definition.

API

In Tracecat, the way we create UDFs is by defining a Python function and decorating it with @registry.register. In Python, we support using both synchronous and asynchronous functions. For example:

Example UDFs
import asyncio
from tracecat.registry import registry

@registry.register(
    description="Adds two numbers.",
    namespace="example",
)
def add(
    lhs: Annotated[int, Field(..., description="The first number")],
    rhs: Annotated[int, Field(..., description="The second number")],
) -> int:
    return lhs + rhs

@registry.register(
    description="Sleeps and subtracts two numbers.",
    namespace="example",
)
async def sleeping_subtract(
    lhs: Annotated[int, Field(..., description="The first number")],
    rhs: Annotated[int, Field(..., description="The second number")],
) -> int:
    await asyncio.sleep(1)
    return lhs - rhs

These UDFs will be registered under the example namespace, and their keys are example.add and example.sleeping_subtract.

The register decorator also accepts arbitrary keyword arguments that are stored in the registry. We plan on using this to further extend the capabilities of UDFs, for example by using a version keyword argument to specify the version of the UDF.

Secrets injection

You can declare a list of secrets in your UDFs using the secrets parameter. You will have to create these secrets on the platform either through the settings/credentials UI or through the API/CLI. The declared secrets are pulled from Tracecat’s secret manager at runtime.

For example, given my_secret with key MY_SECRET_KEY:

Example UDF with a declared secret
@registry.register(
  description="This is a sample UDF with secrets",
  secrets=["my_secret"] # Accepts more than one secret name
)
async def requires_api_key(resource_name: str, value: int) -> str:
    # Call this normally like regular Python code!
    api_key = os.environ["MY_SECRET_KEY"]

    async with httpx.AsyncClient() as client:
        response = await client.post(
          f"https://api.example.com/resource/{resource_name}",
          headers={"Authorization": f"Bearer {api_key}"},
          json={"value": value}
        )
        response.raise_for_status()
    return response.text

For more information on how to create secrets and how Tracecat’s secret manager works, see Secrets.

Schemas

Tracecat performs schema validation on all inputs and outputs of actions.

Schema Generation

  • Tracecat UDFs automatically generate Pydantic models for input and output validation.

Schema validation

Currently all input validation is performed on the server side using the registered UDF Pydantic validators.

UI Generation

  • Motivation: All dyanmic forms and schema validation should be driven by server-side code.
  • UDFs can capture a lot more infromation through the typing.Annotated annotations and registry.register decorator.
  • This information is used to generate extended JSONSchema that can be used to generate UI in our NextJS frontend.

NOTE: This feature is currently a work in progress.