If running in function execution mode, you must register the functions in your compute module to make them callable from elsewhere in Foundry. This page explains two different methods for manually registering a compute module function.
The compute modules SDK makes it easier to register functions by automatically inferring the schema of your function(s). If you are using the compute modules SDK, review the automatic function schema inference section below.
You can manually register a function for a compute module from the Functions page. Select Add function to open the Create function panel:
Compute module function inputs are packaged into a JSON object; each input that you add corresponds to a property on the input object passed into your function. In the example below, function inputs are on the left, and the JSON object passed to the corresponding function is on the right.
example_function_payload.json
Copied!1 2 3 4 5
{ "arg1": "hello", "arg2": 2, "arg3": "1969-07-20" }
com.<namespace>.computemodules.<MyApiName>
and must comply to the following naming rules:Changing the API name will break the consumer code. Only the latest published version of the query is supported.
Once you define a function, you can switch to the Test tab to try invoking the function, and/or select Save to save the function and make it callable from Foundry.
Compute module functions are always registered with version 0.0.0
. If you update the function, the function's version will be overwritten by your changes.
You can also manually define your function schema by sending an HTTP POST request from within your compute module. Typically, you will only need to do this if you are creating your own client. For information on the HTTP request, review our POST function schema documentation.
This endpoint accepts a JSON array as the payload, where each element in the array corresponds to the specification of a function in your compute module. Our Python SDK ↗ provides a good reference on how to assemble this JSON payload.
Below is a table showing the mapping between function input/output types and how those types are serialized over HTTP to a compute module:
Foundry type | Serialsed over HTTP as | Notes |
---|---|---|
Integer | int | |
Byte | string | |
Boolean | boolean | |
Binary | string | |
Date | string | |
Timestamp | int | Milliseconds since epoch |
Decimal | string | |
Float | float | |
Array | array (non-streaming), stream of JSON (streaming) | |
Map | JSON | Key-value store (for exmaple, Python dict , Java Map) |
Struct | JSON | Custom object type |
Compute modules offer a streamlined way to define and register functions, enabling automatic schema inference and integration with Foundry's Compute Module application. This section provides an in-depth look at the automatic registration of functions and advanced usage scenarios, ensuring a smoother development experience.
The imported function schemas will only appear in the Compute Modules interface once your compute module is running and responsive. This means that you must deploy and run your compute module for the functions to be visible and accessible in Foundry. Review our documentation on debugging using replica status for more details.
In your compute module, you can define the schema of a function using a JSON structure directly within your code. This approach offers several benefits:
By making a simple POST
call when your compute module starts up, the module automatically infers the schema from the endpoint call and makes it available as a function in the Computes Modules application. This allows developers to define endpoint schemas once and easily import them into Foundry.
Consider a simple add
function, where inputs are x
and y
(two integers) and the output is a string. The example below shows how to define the JSON schema for this function:
schemas.json
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32
{ "functionName": "add", "inputs": [ { "name": "x", "dataType": { "integer": {}, "type": "integer" }, "required": true, "constraints": [] }, { "name": "y", "dataType": { "integer": {}, "type": "integer" }, "required": true, "constraints": [] } ], "output": { "single": { "dataType": { "string": {}, "type": "string" } }, "type": "single" } }
Once you have defined your JSON schema, send an HTTP POST request in your app.py
file to register it with Foundry:
if __name__ == "__main__":
certPath = os.environ['CONNECTIONS_TO_OTHER_PODS_CA_PATH']
postSchemaUri = os.environ["POST_SCHEMA_URI"]
with open('schemas.json', 'r') as file:
SCHEMAS = json.load(file)
requests.post(
postSchemaUri,
json=SCHEMAS,
headers={"Module-Auth-Token": moduleAuthToken, "Content-Type": "application/json"},
verify=certPath
)
Make sure to handle exceptions and implement proper error logging in a production environment.
Notice that the function adheres to the following constraints:
functionName
that matches the Python function name.Python type | Foundry type | Serialsed over HTTP as | Notes |
---|---|---|---|
int | Integer | int | |
str | Byte | string | |
bool | Boolean | boolean | |
bytes | Binary | string | |
datetime.datetime | Date | string | |
datetime.datetime | Timestamp | int | Milliseconds since epoch |
decimal.Decimal | Decimal | string | |
float | Float | float | |
list | Array | array (non-streaming), stream of JSON (streaming) | |
set | Array | array (non-streaming), stream of JSON (streaming) | |
dict | Map | JSON | Key-value store (for example, Python dict , Java Map) |
class/TypedDict | Struct | JSON | Custom object type |
Iterable | Array | array (non-streaming), stream of JSON (streaming) |
The compute module SDK includes functionality for automatic function discovery. It inspects the defined functions and their input/output types, then converts them into FunctionSpecs
that can be imported as Foundry Functions without modification.
To ensure this feature works seamlessly, you should understand how type inference works within the SDK and how to correctly define input and output types. Review the following considerations:
✅ TypedDict as input type
Copied!1 2 3 4 5 6 7 8 9 10
# app.py from typing import TypedDict from compute_modules.annotations import function class HelloInput(TypedDict): planet: str @function def hello(context, event: HelloInput) -> str: return "Hello " + event["planet"] + "!"
✅ dataclass as input type
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
# app.py from compute_modules.annotations import function from dataclasses import dataclass import datetime import decimal @dataclass class TypedInput: bytes_value: bytes bool_value: bool date_value: datetime.date decimal_value: decimal.Decimal float_value: float int_value: int str_value: str datetime_value: datetime.datetime other_date_value: datetime.datetime @function def typed_function(context, event: TypedInput) -> str: diff = event.other_date_value - event.datetime_value return f"The difference between the provided dates is {diff}"
✅ Regular class with both class AND constructor type hints
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14
# app.py from compute_modules.annotations import function class GoodExample: some_flag: bool some_value: int def __init__(self, some_flag: bool, some_value: int) -> None: self.some_flag = some_flag self.some_value = some_value @function def typed_function(context, event: GoodExample) -> int: return event.some_value
❌ AVOID Python class with no class type hints
Copied!1 2 3 4 5
# app.py # This will raise an exception class BadClassNoTypeHints: def __init__(self, arg1: str, arg2: int): ...
❌ AVOID Python class with Args
in constructor
Copied!1 2 3 4 5 6 7 8
# app.py # This will raise an exception class BadClassArgsInit: arg1: str arg2: int def __init__(self, arg1: str, arg2: int, *args): ...
❌ AVOID Python class with Kwargs
in constructor
Copied!1 2 3 4 5 6 7 8
# app.py # This will raise an exception class BadClassKwargsInit: arg1: str arg2: int def __init__(self, arg1: str, arg2: int, **kwargs): ...
Iterable
type (except dict
). To enable result streaming, change @function
to @function(streaming=True). You can review more details in our [SDK documentation ↗](https://github.com/palantir/python-compute-module?tab=readme-ov-file#advanced-usage-1---streaming-result). To make sure your streaming function is registered correctly, use any
Iterabletype as the return type. Then the output will be registered as Foundry
Array`.If you do not set streaming=True
, the result will be posted as a single JSON blob of the whole iterable. It may throw if your iterable is not able to be serialized in JSON. If you set streaming=True
, the result will be posted as a stream of JSON blobs serialized from each element. Review more in our SDK documentation ↗.
✅ Regular Iterable
as output type
Copied!1 2 3 4 5 6 7 8 9 10 11
# app.py # The outputs will be registered as Foundry Array from compute_modules.annotations import function @function(streaming=True) def get_string_list(context, event) -> list[str]: return [f'string {i}' for i in range(10)] @function(streaming=True) def get_string_set(context, event) -> set[str]: return {'string 1', 'string 2', 'string 3'}
✅ Generator as output type
Copied!1 2 3 4 5 6 7 8 9
# app.py # Generator is Iterable. The output will be registered as Foundry Array from compute_modules.annotations import function import typing @function(streaming=True) def string_generator(context, event) -> typing.Iterable[str]: for i in range(10): yield f'string {i}'
⚠️ Regular Iterable as output type but streaming not enabled
Copied!1 2 3 4 5 6 7
# app.py # This is valid. The output will be registered as Foundry Array, but the result will not be streamed from compute_modules.annotations import function @function def get_string_list(context, event) -> list[str]: return [f'string {i}' for i in range(10)]
❌ Generator as output type but streaming not enabled
Copied!1 2 3 4 5 6 7 8 9 10
# app.py # Generator is not JSON serializable as a whole object. Cannot be used in a non-streaming function since it serializes the whole object # The output type will be registered as Foundry Array, but it will throw when executed from compute_modules.annotations import function import typing @function def string_generator(context, event) -> typing.Iterable[str]: for i in range(10): yield f'string {i}'
Follow the steps below to register your function:
Prerequisites:
Before you begin, ensure that resource generation is enabled in your Typescript code repository:
functions.json
file.enableResourceGeneration
property to true
.To import a compute module function in TypeScript, follow the steps below:
The example below shows how to import and use a compute module function:
// index.ts
import { Function } from "@foundry/functions-api";
// API Name: com.mycustomnamespace.computemodules.Add
import { add } from "@mycustomnamespace/computemodules";
export class MyFunctions {
@Function()
public async myFunction(): Promise<string> {
return await add({ x: 50, y: 50 });
}
}
string
return type, your registered compute module function must return a string
, not a struct
type.async/await
syntax for proper handling.Since TypeScript functions go through the function-executor
, only compute module functions that take less than five minutes will succeed. If the function takes longer than five minutes, it will time out.