Invoking FunctionGraph Event Function using API Calls

This section demonstrates how to call a FunctionGraph implemented in Python using API calls and AK/SK-based authentication.

For details on AK/SK-based authentication, see

• AK/SK-based Authentication

in Identity and Access Management (IAM) API reference.

Prerequisites

For details on prerequisites, see: Prerequisites

Using AK/SK authentication for API calls

Using AK/SK authentication the requests have to be signed with the AK/SK (or for temporal credentials with SecurityAccessKey/SecurityKey/SecurityToken).

For request signing the otc-api-sign-sdk-python can be used.

The SDK provides a method to sign the request with AK/SK or SecurityAccessKey/SecurityKey/SecurityToken.

Additional to the environment variables mentioned in the prerequisites, you also need to set the following environment variables for AK/SK authentication:

• OTC_SDK_AK (Access Key with permission to invoke FunctionGraph)

• OTC_SDK_SK (Secret Key corresponding to the Access Key)

Synchronous invocation using AK/SK

Following sample code demonstrates how to invoke a FunctionGraph event function synchronously using API calls with AK/SK authentication.

See Executing a Function Synchronously in FunctionGraph API reference for more details about synchronous invocation.

First, install dependencies for the sample code:

pip install -r requirements.txt

Code using Python “requests”Code using Python “urllib”

samples-doc/invoke-fg/src/invokeSync_AKSK_requests.py

import json
import os
import sys

from otc_api_sign_core import signer

try:
    from requests import request
except ImportError:
    request = None


def main() -> int:
    if request is None:
        print(
            "Missing dependency: requests. Install it with 'pip install requests'.",
            file=sys.stderr,
        )
        return 1

    region = os.getenv("OTC_SDK_REGION", "eu-de")
    project_id = os.getenv("OTC_SDK_PROJECTID")
    ak = os.getenv("OTC_SDK_AK")
    sk = os.getenv("OTC_SDK_SK")

    missing = [
        name
        for name, value in [
            ("OTC_SDK_PROJECTID", project_id),
            ("OTC_SDK_AK", ak),
            ("OTC_SDK_SK", sk),
        ]
        if not value
    ]
    if missing:
        print(f"Missing required environment variables: {', '.join(missing)}", file=sys.stderr)
        return 1

    fg_endpoint = f"https://functiongraph.{region}.otc.t-systems.com"
    function_name = "python-sample-invoke-function"
    function_version = "latest"
    function_app = "default"

    function_urn = (
        f"urn:fss:{region}:{project_id}:function:{function_app}:{function_name}:{function_version}"
    )

    invoke_uri = f"{fg_endpoint}/v2/{project_id}/fgs/functions/{function_urn}/invocations"
    print("Invoke URI:", invoke_uri)

    x_cff_log_type = "tail"
    x_cff_request_version = "v1"

    body = {
        "key": "Hello T-Cloud Public World - SYNC",
    }
    payload = json.dumps(body)

    method = "POST"
    headers = {
        "Content-Type": "application/json;charset=utf8",
        "Host": f"functiongraph.{region}.otc.t-systems.com",
        "X-Cff-Log-Type": x_cff_log_type,
        "X-CFF-Request-Version": x_cff_request_version,
        "X-Project-Id": project_id,
    }

    sig = signer.Signer()
    sig.Key = ak
    sig.Secret = sk

    signed_request = signer.HttpRequest(method, invoke_uri, headers, payload)
    sig.Sign(signed_request)

    try:
        response = request(
            method,
            invoke_uri,
            headers=signed_request.headers,
            data=payload.encode("utf-8"),
            timeout=30,
        )
        data = response.json()
        print("Status:", response.status_code)
        print("Result:", data.get("result"))
        print("Log:", data.get("log"))
    except Exception as exc:
        print("Error:", exc, file=sys.stderr)
        return 1

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

To execute the sample code, run the following command in the terminal in folder samples-doc/invoke-fg:

python3 src/invokeSync_AKSK_requests.py

In both cases, you should see an output similar to the following in the terminal:

Sample output of synchronous invocation using AK/SK

Result:  {"statusCode":200,"headers":{"Content-Type":"application/json"},"isBase64Encoded":false,"body":"{\"key\": \"Hello T-Cloud Public World - SYNC\"}"}
Log:  2026-03-17T10:13:22Z Start invoke request '960d089c-c812-4fe7-9997-a72f773e4bcb', version: latest
2026-03-17T10:13:22Z Finish invoke request '960d089c-c812-4fe7-9997-a72f773e4bcb', duration: 2.049ms, billing duration: 3ms, memory used: 36.535MB, billing memory: 128MB, cpu used: 0.300U, storage used: 0.039MB

Asynchronous invocation using AK/SK

Following sample code demonstrates how to invoke a FunctionGraph event function asynchronously using API calls with AK/SK authentication.

See Executing a Function Asynchronously in FunctionGraph API reference for more details about asynchronous invocation.

Code using Python “requests”Code using Python “urllib”

samples-doc/invoke-fg/src/invokeASync_AKSK_requests.py

import json
import os
import sys

from otc_api_sign_core import signer

try:
    from requests import request
except ImportError:
    request = None


def main() -> int:
    if request is None:
        print(
            "Missing dependency: requests. Install it with 'pip install requests'.",
            file=sys.stderr,
        )
        return 1

    region = os.getenv("OTC_SDK_REGION", "eu-de")
    project_id = os.getenv("OTC_SDK_PROJECTID")
    ak = os.getenv("OTC_SDK_AK")
    sk = os.getenv("OTC_SDK_SK")

    missing = [
        name
        for name, value in [
            ("OTC_SDK_PROJECTID", project_id),
            ("OTC_SDK_AK", ak),
            ("OTC_SDK_SK", sk),
        ]
        if not value
    ]
    if missing:
        print(f"Missing required environment variables: {', '.join(missing)}", file=sys.stderr)
        return 1

    fg_endpoint = f"https://functiongraph.{region}.otc.t-systems.com"
    function_name = "python-sample-invoke-function"
    function_version = "latest"
    function_app = "default"

    function_urn = (
        f"urn:fss:{region}:{project_id}:function:{function_app}:{function_name}:{function_version}"
    )

    invoke_uri = (
        f"{fg_endpoint}/v2/{project_id}/fgs/functions/{function_urn}/invocations-async"
    )
    print("Invoke URI:", invoke_uri)

    body = {
        "key": "Hello T-Cloud Public World - ASYNC ",
    }
    payload = json.dumps(body)

    method = "POST"
    headers = {
        "Content-Type": "application/json;charset=utf8",
        "Host": f"functiongraph.{region}.otc.t-systems.com",
        "X-Project-Id": project_id,
    }

    sig = signer.Signer()
    sig.Key = ak
    sig.Secret = sk

    signed_request = signer.HttpRequest(method, invoke_uri, headers, payload)
    sig.Sign(signed_request)

    try:
        response = request(
            method,
            invoke_uri,
            headers=signed_request.headers,
            data=payload.encode("utf-8"),
            timeout=30,
        )
        print("Status:", response.status_code)
        print("Response:", response.text)
    except Exception as exc:
        print("Error:", exc, file=sys.stderr)
        return 1

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

To execute the sample code, run the following command in the terminal in folder samples-doc/invoke-fg:

python3 src/invokeASync_AKSK_requests.py

In both cases, you should see an output similar to the following in the terminal:

Sample output of asynchronous invocation using AK/SK

Status: 202
Response:  {"request_id": "3d1a4f5a-b4e2-4429-80c8-3d82d2fe8791"}

last updated: 2026-09-30 08:01:05 UTC - commit: 9655e588c27eff2dd2f5891bc87fe541f6458b7d