Add Akto Guardrails to LiteLLM

[rzeta-10]

Pre-Submission checklist

Please complete all items before asking a LiteLLM maintainer to review your PR

• I have Added testing in the tests/test_litellm/ directory, Adding at least 1 test is a hard requirement - see details
• My PR passes the new test I have added locally
• My PR's scope is as isolated as possible, it only solves 1 specific problem
• I have requested a Greptile review by commenting @greptileai and received a Confidence Score of at least 4/5 before requesting a maintainer review

Type

🆕 New Feature

Changes

• Add Akto Guardrails to LiteLLM

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
litellm	Ready	Preview, Comment	Mar 17, 2026 7:50pm

[CLAassistant]

All committers have signed the CLA.

[greptile-apps]

Greptile Summary

This PR adds a new Akto guardrail integration to LiteLLM using a clean two-entry pattern: akto-validate (pre_call) for synchronous request blocking and akto-ingest (post_call) for fire-and-forget request/response ingestion. The implementation is well-structured and all critical issues surfaced in earlier review rounds have been resolved.

Key changes:

• Core guardrail (akto.py): Implements AktoGuardrail with proper fail-open/fail-closed handling, background_tasks set for GC-safe fire-and-forget tasks, explicit non-200 HTTP error handling, JSON decode error handling, and safe None-guard via result.get("data") or {}.
• Configuration (AktoConfigModel): All six parameters (akto_base_url, akto_api_key, akto_account_id, akto_vxlan_id, unreachable_fallback, guardrail_timeout) are properly defined and forwarded in __init__.py.
• Auto-registration: The module exports guardrail_initializer_registry and guardrail_class_registry which are auto-discovered and merged into the global registries at import time via guardrail_registry.py's module-level update() calls.
• UI: Correctly uses akto.svg for the logo in both guardrailLogoMap and the garden card. The garden preset only configures pre_call mode — users must manually add the post_call ingest entry, which is a known UX limitation.
• Tests: Comprehensive mock-based coverage for all code paths. Note: tests are placed in tests/guardrails_tests/ rather than tests/test_litellm/ as stated in the PR checklist — this is consistent with the existing pattern for guardrail-specific tests but worth confirming with maintainers that this folder is included in the required CI gate.

Confidence Score: 4/5

• Safe to merge with the one minor style fix; all previously flagged critical and major issues have been resolved in this revision.
• All critical bugs from prior review rounds (JSON decode errors, {"data": null} AttributeError, fire-and-forget GC safety, wrong logo extension, missing param forwarding) have been addressed. The implementation follows established LiteLLM guardrail patterns. The only remaining concern is a minor style issue (datetime.now() without explicit UTC) and the test directory placement (tests/guardrails_tests/ vs tests/test_litellm/) which needs confirmation that CI picks it up. No new logic bugs were found.
• No files require special attention beyond the minor datetime style fix in litellm/proxy/guardrails/guardrail_hooks/akto/akto.py.

Important Files Changed

Filename	Overview
litellm/proxy/guardrails/guardrail_hooks/akto/akto.py	Core guardrail implementation with two-entry pattern (pre_call validate, post_call ingest). All critical issues from previous rounds addressed: JSON decode errors are caught, {"data": null} is handled via or {}, fire-and-forget uses background_tasks set for GC safety, and blocked requests raise 403 immediately after scheduling the background ingest task. One minor style concern: datetime.now() used without UTC timezone (functionally equivalent but imprecise).
litellm/proxy/guardrails/guardrail_hooks/akto/init.py	Guardrail initializer properly forwards all config params including akto_account_id and akto_vxlan_id. Exports both guardrail_initializer_registry and guardrail_class_registry which are auto-discovered and merged into the global registries at module import time.
litellm/types/proxy/guardrails/guardrail_hooks/akto.py	Clean config model extending GuardrailConfigModel with all documented fields. No unused imports. All six configurable parameters are properly defined with descriptions, env-var fallback documentation, and appropriate defaults.
tests/guardrails_tests/test_akto_guardrails.py	Comprehensive mock-only test coverage for init, payload building, response parsing, pre/post call paths, fail-open/fail-closed, and helper methods. Correctly uses starlette.exceptions.HTTPException instead of fastapi. Uses double asyncio.sleep(0) to drain fire-and-forget tasks. Tests are placed in tests/guardrails_tests/ instead of the tests/test_litellm/ directory called for by the PR checklist.
litellm/types/guardrails.py	Adds AKTO = "akto" to SupportedGuardrailIntegrations enum and imports + mixes in AktoConfigModel (from canonical location) into LitellmParams. No duplicate class definition — the canonical model is imported correctly.
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts	Adds Akto to GUARDRAIL_PRESETS with mode: "pre_call" only. The integration's two-entry pattern (pre_call validate + post_call ingest) means users who configure via the UI will miss the post_call ingest step — but this was already flagged in previous review rounds.
ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx	Correctly registers akto.svg (matching the actual asset file) in guardrailLogoMap. The previous akto.png typo has been fixed.
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts	Adds Akto card to partner guardrails with correct akto.svg logo, appropriate tags (Security, Safety, Monitoring), and description. Consistent with other partner guardrail entries.

Sequence Diagram

sequenceDiagram
    participant Client
    participant LiteLLM
    participant AktoValidate as AktoGuardrail (pre_call)
    participant AktoIngest as AktoGuardrail (post_call)
    participant AktoAPI as Akto API
    participant LLM

    Client->>LiteLLM: POST /v1/chat/completions

    LiteLLM->>AktoValidate: apply_guardrail(input_type="request")
    AktoValidate->>AktoAPI: POST /api/http-proxy?guardrails=true
    AktoAPI-->>AktoValidate: {data: {guardrailsResult: {Allowed, Reason}}}

    alt Request BLOCKED
        AktoValidate-)AktoAPI: create_task: POST /api/http-proxy?ingest_data=true (statusCode=403)
        AktoValidate->>LiteLLM: raise HTTPException(403)
        LiteLLM-->>Client: 403 Forbidden
    else Request ALLOWED
        AktoValidate-->>LiteLLM: return inputs
        LiteLLM->>LLM: forward request
        LLM-->>LiteLLM: LLM response

        LiteLLM->>AktoIngest: apply_guardrail(input_type="response")
        AktoIngest-)AktoAPI: create_task: POST /api/http-proxy?guardrails=true&ingest_data=true
        AktoIngest-->>LiteLLM: return inputs (immediately)
        LiteLLM-->>Client: 200 OK with LLM response
    end

    Note over AktoAPI: Background ingest runs<br/>asynchronously (fire-and-forget)

Loading

_{Last reviewed commit: ce3f703}

[greptile-apps]

raise_for_status() only raises HTTPStatusError for 4xx/5xx codes. A non-4xx/5xx, non-200 response (e.g. 204 No Content) will bypass the exception and fall through to response.json(), which throws an unhandled JSONDecodeError that bypasses the except (Timeout, httpx.RequestError, httpx.HTTPStatusError) block in apply_guardrail.

Recommend explicitly raising for any non-200 status:

Suggested change

	def handle_guardrail_response(self, response: httpx.Response) -> Tuple[bool, str]:
	"""Parse the Akto response. Raises on non-200 so caller can handle it."""
	if response.status_code != 200:
	verbose_proxy_logger.error(
	"Akto guardrail returned HTTP %d", response.status_code
	)
	response.raise_for_status()
	response_json = response.json()
	verbose_proxy_logger.debug("Akto guardrail response: %s", response_json)
	return self.parse_guardrails_result(response_json)
	def handle_guardrail_response(self, response: httpx.Response) -> Tuple[bool, str]:
	"""Parse the Akto response. Raises on non-200 so caller can handle it."""
	if response.status_code != 200:
	verbose_proxy_logger.error(
	"Akto guardrail returned HTTP %d", response.status_code
	)
	raise httpx.HTTPStatusError(
	f"Akto guardrail returned unexpected status {response.status_code}",
	request=response.request,
	response=response,
	)
	response_json = response.json()
	verbose_proxy_logger.debug("Akto guardrail response: %s", response_json)
	return self.parse_guardrails_result(response_json)

[greptile-apps]

supported_event_hooks advertises GuardrailEventHooks.during_call, but apply_guardrail only accepts input_type: Literal["request", "response"]. There is no code path to handle during_call. This misleads users who configure mode: during_call.

If during_call is not intentionally supported, remove it from the list:

Suggested change

	kwargs["supported_event_hooks"] = [
	GuardrailEventHooks.pre_call,
	GuardrailEventHooks.post_call,
	GuardrailEventHooks.during_call,
	]
	kwargs["supported_event_hooks"] = [
	GuardrailEventHooks.pre_call,
	GuardrailEventHooks.post_call,
	]

[greptile-apps]

GuardrailParamUITypes is imported but never used in this file. Remove the unused import:

Suggested change

	from litellm.types.guardrails import GuardrailParamUITypes
	from pydantic import Field
	from .base import GuardrailConfigModel

[greptile-apps]

FastAPI imports should only appear in proxy/ code. This test file is outside that boundary. Consider using starlette.exceptions.HTTPException instead (which FastAPI is built on) or catching a generic Exception and checking the status code attribute:

Suggested change

	from fastapi import HTTPException
	from starlette.exceptions import HTTPException

Rule Used: What: Do not allow fastapi imports on files outsid... (source)

[greptile-apps]

A second AktoConfigModel is defined here that duplicates the canonical one in litellm/types/proxy/guardrails/guardrail_hooks/akto.py. The two differ in that this one extends BaseModel directly, while the canonical one extends GuardrailConfigModel and includes json_schema_extra and ui_friendly_name().

This creates a maintenance hazard: any new field added to one will not automatically appear in the other. Consider importing the canonical AktoConfigModel from its module and using it directly, following the same pattern as GraySwanGuardrailConfigModel and IBMGuardrailsBaseConfigModel elsewhere in this file.

[greptile-apps]

When a request is blocked, ingest_blocked_request is awaited synchronously before raising HTTPException(400). In the worst case (Akto taking the full timeout), this adds up to guardrail_timeout seconds (default 5 s) of extra latency to every blocked response.

Since ingestion is fire-and-forget observability data (errors are already swallowed inside ingest_blocked_request), consider scheduling it as a background task to return the 400 immediately:

if not allowed:
    import asyncio
    asyncio.create_task(
        self.ingest_blocked_request(
            inputs=inputs,
            request_data=request_data,
            reason=reason,
        )
    )
    raise HTTPException(
        status_code=400,
        detail=reason or "Blocked by Akto Guardrails",
    )

[greptile-apps]

Hardcoded akto_account_id and destIp break multi-tenant deployments

akto_account_id is hardcoded to "1000000" and destIp to "127.0.0.1" for every user of this integration. If Akto assigns different account IDs per installation, all requests sent to a differently-configured Akto instance will be rejected or silently attributed to the wrong account. There is no way for operators to override these values without modifying source code.

At minimum, akto_account_id should be an optional constructor parameter (falling back to "1000000" when not provided) so it can be set via config.yaml. Same rationale applies to destIp if Akto ever needs the real upstream IP.

Suggested change

	"destIp": "127.0.0.1",
	"time": str(int(datetime.now().timestamp() * 1000)),
	"statusCode": str(status_code),
	"type": "HTTP/1.1",
	"status": str(status_code),
	"akto_account_id": "1000000",
	"destIp": "127.0.0.1",
	"time": str(int(datetime.now().timestamp() * 1000)),
	"statusCode": str(status_code),
	"type": "HTTP/1.1",
	"status": str(status_code),
	"akto_account_id": getattr(self, "akto_account_id", "1000000"),
	"akto_vxlan_id": "0",

[greptile-apps]

asyncio.all_tasks() gathers unrelated tasks, causing flaky tests

asyncio.all_tasks() returns every pending task in the running loop, including tasks spawned by pytest-asyncio or other test infrastructure running concurrently. If any such task is slow or fails, this gather will block or raise unexpectedly, making the test non-deterministic.

The intent is to flush the single asyncio.create_task(self.ingest_blocked_request(...)) spawned inside apply_guardrail. A more reliable pattern is to capture that task before it's awaited and await it directly, or use a short asyncio.sleep(0) to yield control once:

# yield control to let the fire-and-forget task run
await asyncio.sleep(0)

This is safer than gathering all running tasks.

[greptile-apps]

Mutating shared request_data dict creates hidden coupling

Setting request_data["_akto_response_ingested"] = True mutates a dict that is shared among all guardrail callbacks for the same request. While this pattern exists in other LiteLLM hooks, it creates invisible coupling: any other code reading request_data (logging, other guardrails, downstream consumers) will now see this internal sentinel key.

If double-ingestion is a real concern (e.g., both pre_call and post_call in block mode), consider tracking this on the instance with a request-scoped identifier (e.g., keyed by request_data.get("litellm_call_id")) instead of polluting the shared dict:

call_id = request_data.get("litellm_call_id", id(request_data))
if call_id in self._ingested_calls:
    return inputs
self._ingested_calls.add(call_id)

[codspeed-hq]

Merging this PR will not alter performance

✅ 16 untouched benchmarks

---

_{Comparing rzeta-10:main (ce3f703) with main (ef9cc33)}

[greptile-apps]

Fire-and-forget task has no strong reference — at risk of GC cancellation

asyncio.create_task(...) returns a Task object but the return value is discarded here. Python's garbage collector can collect an unreferenced task before it finishes running. Per the Python docs: "Save a reference to the result of this function, to avoid a task disappearing mid-execution."

If the GC fires between the task being created and the event loop picking it up, ingest_blocked_request will silently never execute, meaning blocked request details are never sent to Akto with no error surfaced.

The standard fix is to keep a module-level or instance-level set of background tasks:

# at class or module level
_background_tasks: set = set()

# at call site
task = asyncio.create_task(
    self.ingest_blocked_request(
        inputs=inputs,
        request_data=request_data,
        reason=reason,
    )
)
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
raise HTTPException(
    status_code=400,
    detail=reason or "Blocked by Akto Guardrails",
)

[greptile-apps]

Non-200 responses silently ignored in monitor mode

In monitor mode, if the Akto API returns a non-200 status code (e.g., 401, 403, 500), the if response.status_code == 200: guard skips all processing with no log entry. This makes service degradation and misconfiguration (e.g., wrong API key) invisible to operators who rely on logs.

Suggested change

	response = await self.send_request(
	guardrails=True, ingest_data=True, payload=payload
	)
	if response.status_code == 200:
	allowed, reason = self.handle_guardrail_response(response)
	if not allowed:
	verbose_proxy_logger.info(
	"Akto guardrail: response flagged (async mode, logged only): %s",
	reason,
	)
	except Exception as e:
	verbose_proxy_logger.error(
	"Akto guardrail: async post-call error: %s", str(e)
	)
	try:
	response = await self.send_request(
	guardrails=True, ingest_data=True, payload=payload
	)
	if response.status_code == 200:
	allowed, reason = self.handle_guardrail_response(response)
	if not allowed:
	verbose_proxy_logger.info(
	"Akto guardrail: response flagged (async mode, logged only): %s",
	reason,
	)
	else:
	verbose_proxy_logger.warning(
	"Akto guardrail: monitor mode received non-200 status %d",
	response.status_code,
	)
	except Exception as e:
	verbose_proxy_logger.error(
	"Akto guardrail: async post-call error: %s", str(e)
	)

[greptile-apps]

asyncio.all_tasks() gathers unrelated tasks, making the test non-deterministic

asyncio.all_tasks() returns every pending task in the current event loop — including tasks created by pytest-asyncio itself or other concurrently running tests. If any such task is slow or raises, this gather will block or propagate the failure, causing the test to be flaky.

The intended purpose is to flush the single asyncio.create_task(self.ingest_blocked_request(...)) created inside apply_guardrail. A safer pattern is to yield control to the event loop once using asyncio.sleep(0), which is enough to let the fire-and-forget task run without touching unrelated tasks:

Suggested change

	# Await the fire-and-forget background tasks so the mock gets called
	pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
	if pending:
	await asyncio.gather(*pending)
	# yield control once so the fire-and-forget ingest task can run
	await asyncio.sleep(0)

[greptile-apps]

Wrong logo file extension — akto.png doesn't exist

The guardrailLogoMap references akto.png, but the file added to the repository is akto.svg (as used correctly in guardrail_garden_data.ts line 381). This causes the Akto logo to fail to render in any UI component that reads from this map (e.g. the guardrail list/detail view).

Suggested change

	"Akto": `${asset_logos_folder}akto.png`,
	"Akto": `${asset_logos_folder}akto.svg`,

[greptile-apps]

akto_account_id and akto_vxlan_id silently dropped from config

initialize_guardrail constructs AktoGuardrail but never forwards akto_account_id or akto_vxlan_id from litellm_params. Both fields are documented in the config.yaml examples and defined in AktoConfigModel, but any value an operator sets will be silently ignored — the instance always falls back to the environment variable or the hardcoded default.

Add the two missing getattr forwarding calls for akto_account_id and akto_vxlan_id, mirroring how akto_base_url and the other params are already forwarded on lines 16–22.

[greptile-apps]

response.json() raises uncaught ValueError on malformed 200 response

After the explicit non-200 guard at line 324, response.json() is called unconditionally. If Akto returns HTTP 200 with a non-JSON body (e.g., an HTML error page, a plain-text response, or an empty body), response.json() raises json.JSONDecodeError — a subclass of ValueError. This exception is not caught by the except (httpx.RequestError, httpx.HTTPStatusError) block in apply_guardrail, so it propagates all the way up as an unhandled 500, completely bypassing the configured fail_open/fail_closed policy and leaving the client with a confusing internal server error.

result = response.json()
if not isinstance(result, dict):
    return True, ""

Consider wrapping the JSON parse in a try/except and treating a decode error the same way as a network error:

try:
    result = response.json()
except (json.JSONDecodeError, ValueError):
    verbose_proxy_logger.error(
        "Akto returned non-JSON body for status 200: %r",
        response.text[:200],
    )
    raise httpx.RequestError("Akto returned non-JSON body")
if not isinstance(result, dict):
    return True, ""

This ensures apply_guardrail's existing except httpx.RequestError clause picks it up and routes it through handle_unreachable.

[greptile-apps]

UI preset only configures pre_call — post-call ingestion is silently missing

The Akto integration explicitly uses a two-entry pattern (documented in akto.md and in the README-style docstring at the top of akto.py): one pre_call entry for guardrail validation and one post_call entry for request/response ingestion. This preset only creates the pre_call entry.

A user who configures Akto via the guardrail garden UI will end up with:

• ✅ Request blocking (pre-call guardrail check)
• ❌ No data ingestion for allowed traffic (post-call ingest step is absent)

This is inconsistent with every other documented usage example. The ingestion path (akto-ingest) is the primary observability feature of this integration, so its absence would be especially confusing.

If GUARDRAIL_PRESETS only supports single-mode entries today, consider documenting in the preset itself (e.g., via a comment or description field) that the user must add a second post_call entry manually, or look at whether the preset schema can be extended to generate multiple guardrail entries.

[greptile-apps]

AttributeError when Akto returns {"data": null}

result.get("data", {}) returns None when the key is present but its value is None (the default {} is only used when the key is absent). Calling .get(...) on None raises AttributeError.

This exception is not caught by the except (httpx.RequestError, httpx.HTTPStatusError) block in apply_guardrail, so it escapes as an unhandled 500, completely bypassing the configured fail_open/fail_closed policy.

Suggested change

	guardrails_result = result.get("data", {}).get("guardrailsResult", {}) or {}
	if not isinstance(result, dict):
	return True, ""
	data = result.get("data") or {}
	guardrails_result = data.get("guardrailsResult") or {}

[greptile-apps]

UI preset missing post_call ingestion entry

The Akto integration uses an explicit two-entry pattern (documented in akto.md and in the module docstring): akto-validate (pre_call) for guardrail checking and akto-ingest (post_call) for request/response ingestion. The GuardrailPreset interface only supports a single mode: string, so the current preset creates only the pre_call entry.

A user who configures Akto entirely through the guardrail garden UI will end up with request blocking working correctly but zero data ingestion for allowed traffic — the core observability use-case of the Akto integration will silently not function.

Consider one of:

1. Adding a note/description field to the preset that prompts the user to manually add the second post_call entry after applying the preset.
2. Extending GuardrailPreset to support an array of modes and generating both entries when the preset is applied.
3. Documenting the limitation in the preset entry itself.

[greptile-apps]

Single sleep(0) may not be sufficient to fully drain the fire-and-forget task

The fire-and-forget task in apply_guardrail involves a chain of await calls: asyncio.create_task(fire_and_forget_request(...)) → await self.send_request(...) → await self.async_handler.post(...). A single asyncio.sleep(0) yields the event loop once, which is enough to start the task but may not be enough to drive it to completion when there are multiple suspension points.

In practice this works because AsyncMock resolves awaits eagerly in the same event loop tick, but this is an implementation detail of AsyncMock that could change. A more robust approach that documents intent clearly:

# Allow all pending coroutines (the fire-and-forget task chain) to complete
await asyncio.sleep(0)  # schedule task
await asyncio.sleep(0)  # complete send_request

Or alternatively use asyncio.gather(*akto_validate.background_tasks) to drain precisely the tasks the guardrail itself spawned, without touching unrelated loop tasks.

[greptile-apps]

httpx.RequestError constructed with potentially None request

getattr(response, "request", None) can return None for non-httpx.Response objects (e.g., mocks without a request attribute). The httpx.RequestError constructor's request parameter is typed as Request (not Optional[Request]), and while Python doesn't enforce this at runtime, some versions of httpx do validate the type and will raise a TypeError at construction time.

In production the response here is always a real httpx.Response (which always has .request), so the None path won't be hit. But the test at line 308 sets mock_resp.request = MagicMock() explicitly to avoid this, suggesting the defensive getattr(..., None) default could backfire if a future test or edge case omits it. A cleaner approach:

raise httpx.RequestError(
    "Akto returned non-JSON body",
    request=response.request,
) from e

Since handle_guardrail_response is only ever called with a real httpx.Response, response.request is always valid.

[greptile-apps]

Naive datetime used for timestamp generation

datetime.now() returns a timezone-naive datetime. While .timestamp() always returns a UTC epoch value regardless, using a naive datetime is technically ambiguous — if someone reads the intermediate datetime object (e.g., for logging), its timezone is unclear. Prefer the explicit UTC form to make intent clear:

Suggested change

	"time": str(int(datetime.now().timestamp() * 1000)),
	"time": str(int(datetime.now(timezone.utc).timestamp() * 1000)),

This also requires updating the import at the top of the file:

from datetime import datetime, timezone

_{Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!}