SDK Reference¶

Note

Contact your F5 representative for access to evaluate the SDK.

Processor SDK for F5 AI Gateway

exception f5_ai_gateway_sdk.ProcessExecutionError(detail='problem executing processor implementation')¶

Error while a processor executes process per contractual agreement.

Parameters:: detail (str) – Error message

exception f5_ai_gateway_sdk.ProcessorError(status_code, detail, messages=None)¶

Error has occurred during processing of either the input.content or response.content.

Parameters:

status_code (int) – HTTP status code
detail (str) – Error message
messages (List[str]) – Optional list of error messages

class f5_ai_gateway_sdk.Choice(*, message)¶

Represents a choice returned from an upstream.

Parameters:: message (Message) – The Message object associated with the choice.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class f5_ai_gateway_sdk.Message(*, content, role='user')¶

Represents a message with a role and content.

Parameters:

content (str) – The text content of the message.
role (str) – The role of the message, which can be one of the values from MessageRole enum. Default is USER.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class f5_ai_gateway_sdk.MessageRole¶

Enum for the fields that can be required in a processor signature.

Available values:

USER: Represents the user role in the conversation.
SYSTEM: Represents the system role responsible for managing the conversation context.
ASSISTANT: Represents the assistant role, typically used for responses from the AI.
TOOL: Represents an external tool or resource invoked during the conversation.

class f5_ai_gateway_sdk.Processor(name, version, namespace, signature, prompt_class=<class 'f5_ai_gateway_sdk.request_input.RequestInput'>, response_class=<class 'f5_ai_gateway_sdk.response_output.ResponseOutput'>, parameters_class=None, app_details=None)¶

Creates a new instance of a Processor.

Parameters:

name (str) – name of processor (used in API definition)
version (str) – version of processor (used in API definition)
namespace (str) – optional namespace used to identify a group of processors
signature (Signature) – used to validate incoming requests
prompt_class (Type[PROMPT]) – type for class that defines the prompt class for the processor
response_class (Type[RESPONSE]) – type for class that defines the response class for the processor
parameters_class (Type[PARAMS] | None) – type for class that defines the config parameters for the processor
app_details (Dict[str, str | int | float | bool | None | List[JSONTypes] | Dict[str, JSONTypes]] | None) – optional dictionary of application details (e.g. version, description)

Example:

class ExampleProcessor(Processor):
    def __init__(self):
        super().__init__(
            name="example-processor",
            version="v1",
            namespace="tutorial",
            signature = BOTH_SIGNATURE
        )

choose_span(name)¶

This method chooses the current span to use for the processor. If the current span is not recording, then it will return the current span which means that we don’t start a new span that starts recording when we don’t want it to. If the current span is recording, then we start a new span which automatically becomes a sub-span of the current span.

Parameters:: name (str) – name of span
Returns:: current non-recording span or a new span

async handle_request(request)¶

This method handles all requests from the Starlette route and handles any errors that may occur during the processor execution.

Parameters:: request (Request) – HTTP request in the AI Gateway Processor format
Returns:: HTTP response in the AI Gateway Processor format
Return type:: Response

abstract process(prompt, response, metadata, parameters, request)¶

This abstract method is for implementors of the processor to define with their own custom logic. Errors should be raised as a subclass of ProcessorError. The return value should be a Result object.

Parameters:

prompt (PROMPT | None) – Optional prompt content to be evaluated
response (RESPONSE | None) – Optional response content to be evaluated
metadata (dict[str, Union[str, int, float, bool, NoneType, List[Union[str, int, float, bool, NoneType, List[ForwardRef('JSONTypes')], Dict[str, ForwardRef('JSONTypes')]]], Dict[str, Union[str, int, float, bool, NoneType, List[ForwardRef('JSONTypes')], Dict[str, ForwardRef('JSONTypes')]]]]]) – Prompt/Response metadata
parameters (PARAMS) – Processor parameters - configuration sent by gateway
request (Request) – Original HTTP request

Returns:

Result object indicating if the request was rejected, accepted, or modified

Return type:

Result

Example:

def process(self, prompt, response, metadata, parameters, request):
    target_found = False
    target_str = "target"

    if response:
        target_found = any(
            target_str in choice.message.content for choice in response.choices
        )
    elif prompt:
        target_found = any(target_str in message.content for message in prompt.messages)

    result = Metadata({"target_found": target_found})

    return Result(processor_result=result)

app_details¶: Runtime details of the application using the processor

methods¶: Supported HTTP methods for the processor API

name¶: Name of the processor (without spaces)

namespace¶: Namespace of the processor (used to group processors)

namespaced_path¶: Namespaced path for the processor API

parameters_class¶: Class that defines the configuration parameters for the processor

path¶: Path for the processor API that can be extended for subcommands

prompt_type¶: Type of the prompt content

response_type¶: Type of the response content

signature¶: Signature of the processor (used to verify requests)

version¶: Version of the processor (independent of API version)

class f5_ai_gateway_sdk.ProcessorRoutes(processors, root_path=None)¶

Create a new ProcessorRoutes object

Parameters:

processors (Iterable[Processor]) – processors to load as routes
root_path (str | None) – root path of processor server

Example:

routes=ProcessorRoutes([MyProcessor()])

# Create a Starlette app
app = Starlette(routes=routes)

# Or create a FastAPI app
app = FastAPI(routes=routes)

class f5_ai_gateway_sdk.RequestInput(*, messages)¶

Represents a collection of Message objects.

Parameters:: messages (List[Message]) – A list of Message objects representing the input messages.

Example:

{
    "messages": [
        {
            "content": "What is the capital of France?",
            "role": "user"
        },
        {
            "content": "Only answer questions about geography",
            "role": "system"
        }
    ]
}

concatenate(roles=None)¶

Concatenates the message contents, filtered by roles if provided.

Parameters:: roles (List[MessageRole] | None) – An optional list of roles to filter the messages by
Returns:: A concatenated string of message contents.
Return type:: str

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class f5_ai_gateway_sdk.ResponseOutput(*, choices)¶

Represents a collection of Message objects.

Parameters:: choices (List[Choice]) – A list of Choice objects.

Example:

{
    "choices": [
        {
            "message:"
                {
                    "content": "The capital of France is Paris.",
                    "role": "assistant",
                }
        },
    ]
}

choices: List[Choice]¶: A list of Choice objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class f5_ai_gateway_sdk.Result(modified_prompt=None, modified_response=None, metadata=None, processor_result=None, tags=None, rejected=False, modified=False)¶

Parameters:

modified_prompt (RequestInput | None) – rewritten or modified prompt to be returned in the HTTP response
modified_response (ResponseOutput | None) – rewritten or modified response to be returned in the HTTP response
metadata (dict[str, Union[str, int, float, bool, NoneType, List[ForwardRef('JSONTypes')], Dict[str, ForwardRef('JSONTypes')]]] | None) – root metadata containing pre-defined fields
processor_result (dict[str, Union[str, int, float, bool, NoneType, List[ForwardRef('JSONTypes')], Dict[str, ForwardRef('JSONTypes')]]] | None) – free-form metadata resulting from process execution
tags (Tags | None) – used for annotating a transaction with notable properties
rejected (bool) – flag indicating that the prompt/metadata submitted is unacceptable
modified (bool) – flag indicating that the prompt has been modified

to_response()¶

Converts the Result object to a Starlette Response object, with the following rules:: If the result is rejected, a 422 status code is returned. If the prompt was modified, the modified prompt is sent in the body along with the metadata, and a 200 status code is returned. If the prompt was not modified and a processor result (additional metadata) is set, the processor result is sent in the body, and a 200 status code is returned. If the prompt was not modified and no processor result is set, no body is sent, and a 204 status code is returned.

Returns:: Starlette Response object
Return type:: Response

class f5_ai_gateway_sdk.Signature(required=None, optional=None)¶

Signature for a processor that defines the required input and response fields for a processor. These values are used to inform the gateway of what inputs are acceptable.

The following predefined Signatures cover most common use cases.

INPUT_ONLY_SIGNATURE
RESPONSE_ONLY_SIGNATURE
RESPONSE_AND_PROMPT_SIGNATURE
BOTH_SIGNATURE
BOTH_RESPONSE_PROMPT_SIGNATURE

Parameters:

required (set[f5_ai_gateway_sdk.signature.SignatureField] | frozenset[f5_ai_gateway_sdk.signature.SignatureField] | None) – A set of SignatureField objects representing the required input fields.
optional (set[f5_ai_gateway_sdk.signature.SignatureField] | frozenset[f5_ai_gateway_sdk.signature.SignatureField] | None) – A set of SignatureField objects representing the optional input fields.

supports_input()¶

Returns True if the signature requires works with input requests.

Returns:: bool: Whether the signature works with input requests.

Return type:: bool

supports_response()¶

Returns True if the signature requires works with response requests.

Returns:: bool: Whether the signature works with response requests.

Return type:: bool

to_list()¶

Return a list of dicts that represent the fields required by this signature

Return type:: List

class f5_ai_gateway_sdk.SignatureField(value)¶

Enum for the fields that can be required in a processor signature.

Available values:

INPUT: The input stage.
RESPONSE: The response stage.

class f5_ai_gateway_sdk.Tags(initial_tags=None)¶

Initialize the Tags object with an optional initial dictionary.

Example:

tags = Tags({"topics_detected": ["security", "ai"]})

Parameters:: initial_tags (Dict[str, List[str]] | None) – An optional dictionary of key-value pairs with list of strings.

add_tag(key, *values)¶

Add a tag to the given key.

Parameters:

key (str) – The key to which the tag will be added.
values (str) – The values of the tags to be added.

Return type:

None

get_all_tags()¶

Get the entire tags dictionary.

Return type:: str | int | float | bool | None | List[str | int | float | bool | None | List[JSONTypes] | Dict[str, JSONTypes]] | Dict[str, str | int | float | bool | None | List[JSONTypes] | Dict[str, JSONTypes]]

get_tags(key)¶

Get all tags associated with the given key.

Parameters:: key (str) – The key to retrieve tags for.
Returns:: A set of all tags associated with the given key. If no tags are found, an empty list is returned.
Return type:: List[str]

remove_key(key)¶

Remove a key from the tags object.

Parameters:: key (str) – The key to be removed.
Return type:: None

remove_tag(key, value)¶

Remove a tag from the given key.

Parameters:

key (str) – The key from which the tag will be removed.
value (str) – The value of the tag that will be removed.

Return type:

None

to_response()¶

Convert Tags with are string dict with sets to dicts with lists for JSON serialization :return: A dictionary where each key has a list of strings as its value.

Return type:: str | int | float | bool | None | List[str | int | float | bool | None | List[JSONTypes] | Dict[str, JSONTypes]] | Dict[str, str | int | float | bool | None | List[JSONTypes] | Dict[str, JSONTypes]]

Previous Next