User-Defined Functions

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:

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:

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