SDK Reference¶

Processor SDK for F5 AI Gateway

This source code is licensed under the Apache License Version 2.0 found in the LICENSE file in the root directory of this source tree.

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.Parameters(*, annotate=True, modify=False, reject=False)¶

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

annotate (bool) –
modify (bool) –
reject (bool) –

otel_attributes(key_prefix='parameters.')¶

Return OpenTelemetry attributes to be sent with the processor span.

Parameters:: key_prefix (str) –
Return type:: Iterator[tuple[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]]

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

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, JsonValue] | 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

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

DEPRECATED: Implement ‘process_input’ and/or ‘process_response’ instead.

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 or Reject object.

Parameters:

prompt (PROMPT | None) – Optional prompt content to be evaluated
response (RESPONSE | None) – Optional response content to be evaluated
metadata (dict[str, JsonValue]) – 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 | Reject

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)

process_input(prompt, metadata, parameters, request)¶

Parameters:

prompt (PROMPT) – Optional prompt content to be evaluated
metadata (dict[str, JsonValue]) – 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 | Reject

Example:

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

    target_found = any(target_str in message.content for message in prompt.messages)
    result = Metadata({"target_found": target_found})

    return Result(processor_result=result)

process_response(prompt, response, metadata, parameters, request)¶

Parameters:

prompt (PROMPT | None) – Optional prompt content to be evaluated
response (RESPONSE) – Optional response content to be evaluated
metadata (dict[str, JsonValue]) – 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 | Reject

Example:

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

    target_found = any(
        target_str in choice.message.content for choice in response.choices
    )
    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.Reject(*, code, detail, metadata={}, tags=Tags())¶

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

code (RejectCode) –
detail (str) –
metadata (dict[str, JsonValue]) –
tags (Tags) –

is_empty()¶

Compatability with Result(), always false due to required fields

Return type:: bool

to_multipart_field()¶

Convert to response object

Return type:: MultipartResponseField

to_response()¶

Return Reject as Response object

Return type:: Response

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

class f5_ai_gateway_sdk.RejectCode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶: Error codes available for rejects

class f5_ai_gateway_sdk.RequestInput(*, messages)¶

Represents a collection of Message objects.

Parameters:: messages (list[f5_ai_gateway_sdk.request_input.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[f5_ai_gateway_sdk.request_input.MessageRole] | None) – An optional list of roles to filter the messages by
Returns:: A concatenated string of message contents.
Return type:: str

hash()¶

Return hash of model

Return type:: int

to_multipart_field()¶

Convert to response object

Return type:: MultipartResponseField

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[f5_ai_gateway_sdk.response_output.Choice]) – A list of Choice objects.

Example:

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

hash()¶

Return hash of model

Return type:: int

to_multipart_field()¶

Convert to response object

Return type:: MultipartResponseField

choices: list[f5_ai_gateway_sdk.response_output.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=Tags())¶

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

modified_prompt (RequestInput | None) –
modified_response (ResponseOutput | None) –
metadata (dict[str, JsonValue] | None) –
processor_result (dict[str, JsonValue] | None) –
tags (Tags) –

check_prompt_or_response()¶

Check that a processor is not modifying prompt and response together. If response is present the request is in a responseStage, so modifying the prompt will have no impact.

Return type:: Self

to_response()¶

Converts the Result object to a Starlette Response object, with the following rules:: 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 or metadata is set, no body is sent, and a 204 status code is returned.

Returns:: Starlette Response object
Return type:: Response

validate_allowed(processor_name, annotate, modify)¶

Validate modifications and annotations based on parameters

Parameters:

processor_name (str) –
annotate (bool) –
modify (bool) –

property is_empty: bool¶: Check if anything has been set on instance

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

property modified: bool¶: Check if a modification exists

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, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶

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:: JsonValue

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]

model_post_init(context, /)¶

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

Parameters:

self (BaseModel) –
context (Any) –

Return type:

None

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:: JsonValue

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

Previous Next