F5 AI Gateway Python SDK Quickstart¶

About processors¶

There are three types of alterations which a processor may make on a request.

Annotation
Rejection
Modification

A processor can only make these types of alterations when a specific parameter is provided.

If there is an attempt to make a modification without the required parameter, the SDK will drop the modification, and it will log a warning. For example:

WARNING:root:CatClassifierProcessor tried to annotate request with tags when parameters.annotate was set to false, tags will be dropped

Annotation¶

A processor can add tags to a request which can be used by AI Gateway to route the request. This is controlled by the parameter annotate, which defaults to true.

Rejection¶

A processor can reject a request which causes a rejection message to be returned to the client. This is controlled by the parameter reject, which defaults to false.

Modification¶

A processor can modify the prompt that is sent by a client or modify the response from an upstream model. This is controlled by the parameter modify, which defaults to false.

Before you begin¶

To follow the steps in this guide you will need:

A Python 3.11+ environment with pip or an alternative installed
An ASGI server

In this example we will use uvicorn as our ASGI server. Install it first if you don’t have it already or adjust the commands to use the server of your choice.

Note

You may want to use a Python virtual environment to manage your dependencies. Below is an example of setting up and activating a virtual environment.

python3 -m venv .venv
source .venv/bin/activate

pip install uvicorn

Install the SDK¶

The F5 AI Gateway Python SDK is designed to handle requests and responses between AI Gateway and your application, allowing you to focus on developing your processor logic.

pip install git+https://github.com/nginxinc/f5-ai-gateway-sdk-py

You can also install using one of the .whl files that are available from the releases page.

pip install f5_ai_gateway_sdk-<version>-py3-none-any.whl

Create a basic processor which annotates a request¶

In this tutorial we create a simple processor that inspects the prompt and response for any mentions of a ‘cat’ and tags it accordingly. Tags can be used in the AI Gateway configuration to route to a specific upstream LLM or run a particular set of processors.

Create a file called cat_classifier.py with the following contents.

from f5_ai_gateway_sdk.parameters import Parameters
from f5_ai_gateway_sdk.processor import Processor
from f5_ai_gateway_sdk.processor_routes import ProcessorRoutes
from f5_ai_gateway_sdk.request_input import Message, MessageRole
from f5_ai_gateway_sdk.result import Result, Reject, RejectCode
from f5_ai_gateway_sdk.signature import BOTH_SIGNATURE, INPUT_ONLY_SIGNATURE
from f5_ai_gateway_sdk.tags import Tags
from f5_ai_gateway_sdk.type_hints import Metadata
from starlette.applications import Starlette


class CatClassifierProcessor(Processor):
    """
    A simple processor which adds a tag when a cat is found in the prompt or response
    """

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=INPUT_ONLY_SIGNATURE,
        )

    def process_input(self, prompt, metadata, parameters, request):
        my_tags = Tags()
        cat_found = any("cat" in message.content for message in prompt.messages)

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

        if parameters.annotate and cat_found:
            my_tags.add_tag("animals-found", "cat")

        return Result(processor_result=result, tags=my_tags)


app = Starlette(
    routes=ProcessorRoutes([CatClassifierProcessor()]),
)

Note

This example contains all of the imports that will be needed for this tutorial so your editor may warn about unused imports.

Run the processor locally¶

Run the processor locally using uvicorn; this will start a server on port 9999. Ensure that you run this command from the directory where your cat_classifier.py file is located.

python -m uvicorn cat_classifier:app --host 127.0.0.1 --port 9999 --reload

The --reload flag will automatically restart the server when you make changes to your code.

We can check if the server is running by sending a curl request to the signature endpoint of the processor.

Request

curl -i http://localhost:9999/api/v1/signature/tutorial/cat-classifier

Response

{
    "fields": [
        {
            "type": "input.messages",
            "required": true
        }
    ],
    "parameters": {
        "description": "Default empty parameters class for processors that do not require parameters.\nThis class should not be inherited from.",
        "properties": {
            "annotate": {
                "default": true,
                "description": "Whether the processor can annotate the input with tags.",
                "title": "Annotate",
                "type": "boolean"
            },
            "modify": {
                "default": false,
                "description": "Whether the processor can modify the input.",
                "title": "Modify",
                "type": "boolean"
            },
            "reject": {
                "default": false,
                "description": "Whether the processor can reject requests.",
                "title": "Reject",
                "type": "boolean"
            }
        },
        "title": "Default Parameters",
        "type": "object"
    }
}

In the above endpoint, tutorial is the namespace of your processor, and cat-classifier is the name of your processor.

The response should be similar to the following example and contains information that is used by AI Gateway to send correctly formatted requests to the processor:

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious cats gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Fri, 22 Nov 2024 16:33:10 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6"
Transfer-Encoding: chunked

--3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "example:cat-classifier", "processor_version": "v1", "processor_result": {"cat_found": true}, "tags": {"animals-found": ["cat"]}}
--3zWBuRUjyupW5lUbCEITz3riZBm2dPztxregsT4fMZJBHjmoEDYUq5fqAXGCKHr6--

Use the processor with AI Gateway¶

To configure this processor in AI Gateway add the following to your aigw.yml.

processors:
  - name: cat-classifier
    type: external
    config:
      endpoint: "localhost:9999"
      namespace: tutorial
      version: 1

See the configure processors page for more details.

How to reject a request¶

Processors can be configured on a per-request basis by setting the params object in either the processors or steps sections of the AI Gateway configuration file.

Let’s use our default parameters to reject any requests which don’t contain cats. Update the end of the process function in cat_classifier.py to return a Reject object if the condition matches:

        if parameters.reject and not cat_found:
            return Reject(
                code=RejectCode.POLICY_VIOLATION, detail="No cat found in request"
            )

        return Result(processor_result=result, tags=my_tags)

We can now make a request to the processor with the reject parameter set to true:

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={"reject":true};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Fri, 16 May 2025 10:41:17 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="zPn39CWJ76o4UJ3xpp54OkyGVh0zqpnjB9138jOjWLRZjyxlicTwIHgggTqWAZes"
Transfer-Encoding: chunked

--zPn39CWJ76o4UJ3xpp54OkyGVh0zqpnjB9138jOjWLRZjyxlicTwIHgggTqWAZes
Content-Disposition: form-data; name="reject"
Content-Type: application/json

{"code":"AIGW_POLICY_VIOLATION","detail":"No cat found in request"}
--zPn39CWJ76o4UJ3xpp54OkyGVh0zqpnjB9138jOjWLRZjyxlicTwIHgggTqWAZes
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "tutorial:cat-classifier", "processor_version": "v1"}
--zPn39CWJ76o4UJ3xpp54OkyGVh0zqpnjB9138jOjWLRZjyxlicTwIHgggTqWAZes--

Modify a request¶

Processors also have the ability to make modifications to the prompt or response before it is routed to the next stage. Make the following changes to add a system message to the request if the processor is used in an input stage.

Additional parameters can be added to the processor by defining a class that extends f5_ai_gateway_sdk.parameters.Parameters. Add the following after the imports section at the top of the cat_classifier.py file:

class CatClassifierParameters(Parameters):
    enforce_cat_message: str = "Remember to talk about cats in your response"

We then need to update the __init__ function for the processor to specify this class.

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=INPUT_ONLY_SIGNATURE,
            parameters_class=CatClassifierParameters
        )

If we make a request to the signature endpoint, we can see the newly added parameter.

Request

curl -i http://localhost:9999/api/v1/signature/tutorial/cat-classifier

Response

{
    "fields": [
        {
            "type": "input.messages",
            "required": true
        }
    ],
    "parameters": {
        "properties": {
            "annotate": {
                "default": true,
                "description": "Whether the processor can annotate the input with tags.",
                "title": "Annotate",
                "type": "boolean"
            },
            "modify": {
                "default": false,
                "description": "Whether the processor can modify the input.",
                "title": "Modify",
                "type": "boolean"
            },
            "reject": {
                "default": false,
                "description": "Whether the processor can reject requests.",
                "title": "Reject",
                "type": "boolean"
            },
            "enforce_cat_message": {
                "default": "Remember to talk about cats in your response",
                "title": "Enforce Cat Message",
                "type": "string"
            }
        },
        "title": "CatClassifierParameters",
        "type": "object"
    }
}

Note

Due to the CatClassifierParameters used in this example being a subclass of Parameters, it automatically has access to the common parameters of annotate, reject, and modify.

Add the following to the process function before the return statement, and include modified_prompt=prompt in the Result

        if parameters.modify:
            prompt.messages.append(
                Message(
                    content=parameters.enforce_cat_message,
                    role=MessageRole.SYSTEM,
                )
            )

        return Result(processor_result=result, tags=my_tags, modified_prompt=prompt)

Verify that the system message has been added to the request.

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'input.parameters={"modify":true,"enforce_cat_message":"Remember to talk about how cute cats are in your response"};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'input.messages={
    "messages": [
        {
        "content": "Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Thu, 12 Dec 2024 16:03:26 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy"
Transfer-Encoding: chunked

--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy
Content-Disposition: form-data; name="input.messages"
Content-Type: text/plain;charset=utf-8

{"messages":[{"content":"Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows.","role":"user"},{"content":"Remember to talk about how cute cats are in your response","role":"system"}]}
--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "tutorial:cat-classifier", "processor_version": "v1", "processor_result": {"cat_found": false}}
--iBDsmhfPKiowYNQTb9bbfPT76c26q3y8Zj9o7qFHrgDIsIlLqDQi5E5bQaqjIkLy--

Act on the response from the upstream LLM¶

Until now our processor used the process_input function to perform operations on requests before they reach the upstream LLM. We can also perform similar actions on the LLM response before it is returned to the client.

Let’s update our __init__ function to specify that we support processing the response by changing the signature.

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=BOTH_SIGNATURE,
            parameters_class=CatClassifierParameters,
        )

After making this change the processor starts reporting an error.

TypeError: Cannot create concrete class CatClassifierProcessor. Provided Signature supports response but 'process_response' is not implemented.

Let’s implement a basic process_response function to satisfy this requirement. LLMs do not always respond to instructions in the way we expect, so we can reject any responses which ignore the instruction we added in the Modify a request section to mention how cute cats are and does nothing if reject: true is not set.

    def process_response(self, prompt, response, metadata, parameters, request):
        cute_found = any(
            "cute" in choice.message.content for choice in response.choices
        )

        if parameters.reject and not cute_found:
            return Reject(
                code=RejectCode.POLICY_VIOLATION,
                detail="Upstream did not follow instructions",
            )

        return Result()

In this request instead of using input.parameters and input.messages we provide response.parameters and response.choices. The format of input.parameters and response.parameters are the same. However while input.messages contains a list of message objects, response.choices contain a list of choice objects which each contain a message.

Request

curl -i -X POST http://localhost:9999/api/v1/execute/tutorial/cat-classifier \
    -H "Content-Type: multipart/form-data" \
    --form 'response.parameters={"reject":true};type=application/json' \
    --form 'metadata={};type=application/json' \
    --form 'response.choices={
    "choices": [
      {
        "message": {
          "content": "Curious dogs gracefully roam through cozy homes, chasing shadows and basking in warm sunlit windows."
        }
      }
    ]
    };type=application/json'

Response

HTTP/1.1 200 OK
date: Fri, 16 May 2025 14:29:38 GMT
server: uvicorn
content-type: multipart/form-data;charset=utf-8;boundary="Z5pDkuwlewP1uy1QocdQual50l156Y2ceOlM3en5tMlG3NNvisCVlF2loDmosSXw"
Transfer-Encoding: chunked

--Z5pDkuwlewP1uy1QocdQual50l156Y2ceOlM3en5tMlG3NNvisCVlF2loDmosSXw
Content-Disposition: form-data; name="reject"
Content-Type: application/json

{"code":"AIGW_POLICY_VIOLATION","detail":"Upstream did not follow instructions"}
--Z5pDkuwlewP1uy1QocdQual50l156Y2ceOlM3en5tMlG3NNvisCVlF2loDmosSXw
Content-Disposition: form-data; name="metadata"
Content-Type: application/json

{"processor_id": "tutorial:cat-classifier", "processor_version": "v1"}
--Z5pDkuwlewP1uy1QocdQual50l156Y2ceOlM3en5tMlG3NNvisCVlF2loDmosSXw--

Completed tutorial processor¶

from f5_ai_gateway_sdk.parameters import Parameters
from f5_ai_gateway_sdk.processor import Processor
from f5_ai_gateway_sdk.processor_routes import ProcessorRoutes
from f5_ai_gateway_sdk.request_input import Message, MessageRole
from f5_ai_gateway_sdk.result import Result, Reject, RejectCode
from f5_ai_gateway_sdk.signature import BOTH_SIGNATURE, INPUT_ONLY_SIGNATURE
from f5_ai_gateway_sdk.tags import Tags
from f5_ai_gateway_sdk.type_hints import Metadata
from starlette.applications import Starlette


class CatClassifierParameters(Parameters):
    enforce_cat_message: str = "Remember to talk about cats in your response"


class CatClassifierProcessor(Processor):
    """
    A simple processor which adds a tag when a cat is found in the prompt or response
    """

    def __init__(self):
        super().__init__(
            name="cat-classifier",
            version="v1",
            namespace="tutorial",
            signature=BOTH_SIGNATURE,
            parameters_class=CatClassifierParameters,
        )

    def process_input(self, prompt, metadata, parameters, request):
        my_tags = Tags()
        cat_found = any("cat" in message.content for message in prompt.messages)

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

        if parameters.annotate and cat_found:
            my_tags.add_tag("animals-found", "cat")

        if parameters.reject and not cat_found:
            return Reject(
                code=RejectCode.POLICY_VIOLATION, detail="No cat found in request"
            )

        if parameters.modify:
            prompt.messages.append(
                Message(
                    content=parameters.enforce_cat_message,
                    role=MessageRole.SYSTEM,
                )
            )

        return Result(processor_result=result, tags=my_tags, modified_prompt=prompt)

    def process_response(self, prompt, response, metadata, parameters, request):
        cute_found = any(
            "cute" in choice.message.content for choice in response.choices
        )

        if parameters.reject and not cute_found:
            return Reject(
                code=RejectCode.POLICY_VIOLATION,
                detail="Upstream did not follow instructions",
            )

        return Result()


app = Starlette(
    routes=ProcessorRoutes([CatClassifierProcessor()]),
)

Previous Next