FortiAIGate client¶
FortiAIGate ¶
FortiAIGate 8.x client.
FortiAIGate(url, username, password) logs in on the first call (the
login name is the account's EMAIL, or admin for the built-in admin).
FortiAIGate(url, token=...) uses a bearer token instead. verify is
True (system trust store), a CA bundle path (trust a self-signed appliance
certificate), or False (no verification; lab use only). gateway_url is
the data plane's base when it differs from the management URL.
rate_limit_wait is how many seconds in total to spend waiting out HTTP
429s before raising RateLimited (0, the default, raises at once).
Transport failures are retried with backoff (1, 2, 4, ... s): a request that was never answered is retried for every method; a connection dropped AFTER the request went out is retried only for GET, since a mutation may already have landed. HTTP errors are never retried, except one re-login on 401 and one fresh CSRF token on a CSRF 403.
role
property
¶
The login's org role, read once: org_admin, super_admin,
org_member, or the effective workspace_admin/workspace_member.
org_scope
property
¶
True for an organization administrator, whose calls without ws_id
use the /api/v1/admin paths; False for a workspace login.
from_env
classmethod
¶
from_env(env: Mapping[str, str] | None = None, **kw: Any) -> FortiAIGate
FAIG_URL plus FAIG_TOKEN or FAIG_USERNAME/FAIG_PASSWORD. FAIG_CA_BUNDLE trusts a CA file; FAIG_VERIFY_TLS=false turns verification off; FAIG_WORKSPACE sets the default workspace.
Reads os.environ unless env is given, such as
pyfortiaigate.settings.load_settings() (the environment plus your
settings file).
login ¶
Log in now (the client otherwise logs in on its first call).
request ¶
Call the management API. path is /api/v1/... (or /v1/...).
Returns the decoded JSON, or None for an empty body.
default_workspace ¶
The workspace a call without ws_id goes to: the workspace
given to the client (name or id), else, for a workspace login, its only
membership. None for an org admin with no workspace (admin paths).
A workspace login in several workspaces must name one.
invite ¶
invite(email: str, role: str = ROLE_MEMBER) -> Invitation
Invite a user. No email is sent: accept it with accept_invite
or hand over invitePath. A second invite while one is pending is
refused (ValidationError); revoke the first.
invitation_by_email ¶
invitation_by_email(email: str) -> Invitation | None
The pending invitation for an address, if any (case-insensitive).
accept_invite ¶
accept_invite(invitation_id: str, email: str, password: str, display_name: str = '') -> AcceptedInvite
Accept an invitation, creating the account with this password. Unauthenticated in the product, so sent with no session cookie.
make_workspace_admin ¶
The user update REPLACES the user's workspace list, so the existing memberships are kept and this one added. Returns the updated user.
resolve_workspace ¶
resolve_workspace(ws: WsRef) -> str
A workspace's id from its id, its name, or its record. Every method
that takes ws_id accepts all three, so faig.guardrails("team-a")
works. Names are looked up once and cached (refreshed on a miss); an
org admin reads every workspace, anyone else the ones they belong to.
my_workspaces ¶
my_workspaces() -> list[MyWorkspace]
The workspaces the logged-in user is a member of.
create_workspace ¶
create_workspace(name: str, admin_ids: Iterable[str], provider_ids: Iterable[str], description: str = '') -> Workspace
admin_ids must be EXISTING user ids (a pending invitation is not one).
workspace ¶
workspace(ws_id: WsRef) -> WorkspaceSettings
A workspace's settings: allowedModels, allowedPolicies,
allowedGuardrails, budgets and member defaults.
workspace_analytics ¶
workspace_analytics(ws_id: WsRef, frm: Day | None = None, to: Day | None = None, limit: int = 10) -> Analytics
Guardrail analytics for a workspace; see analytics.
resolve_detector ¶
A detector's uuid from its uuid, name (custom or built-in) or record. Every method that takes a detector accepts all three.
resolve_guardrail ¶
A guardrail's id from its id, name or record.
resolve_policy ¶
A policy's id from its id, name or record.
resolve_virtual_key ¶
A virtual key's id from its id, name, record or faig_... secret.
detectors ¶
Custom detectors in a workspace, or (no ws_id) every custom detector.
builtin_detectors ¶
builtin_detectors(ws_id: WsRef | None = None) -> list[BuiltinDetector]
Built-in detectors, identified by a slug id; a guardrail can use either
kind. ws_id uses the workspace path, which a workspace member can call.
create_detector ¶
See regex_detector_body, dlp_detector_body and ai_detector_body.
A name already used in the workspace is a Conflict.
delete_detector ¶
Delete a workspace detector (ws_id) or, as org admin, any detector.
playground ¶
playground(text: str, body: Mapping[str, Any], ws_id: WsRef | None = None) -> DetectorTest
Scan text with a detector BODY that need not be saved (the
console's Playground): build it with regex_detector_body,
dlp_detector_body or ai_detector_body.
Builds without the unsaved-detector endpoint (it answers 404/405) can only
test a SAVED detector. There a regex body is matched locally, as the
console's own Playground does, and the result carries
checkedLocally: True; a DLP or AI body raises FaigError.
test_detector ¶
test_detector(ws_id: WsRef, uuid: Ref, text: str) -> DetectorTest
Scan text with a SAVED detector: matched, finalDisposition,
score, spans. For an unsaved body use playground.
ensure_detector ¶
Make the workspace's detector named body["name"] match body:
create it if missing, update it if any field in body differs, leave
it alone otherwise. approve (org admin) walks a new or still-draft
detector to active. Safe to re-run.
guardrails ¶
A workspace's guardrails, or (no ws_id) the org's.
guardrail ¶
One guardrail, by the workspace path (ws_id) or the org-admin path.
create_guardrail ¶
create_guardrail(ws_id: WsRef, name: str, detector_ids: Sequence[Ref], description: str = '', *, deny_suspicious_characters: bool = False) -> Guardrail
detector_ids may be detector names (custom or built-in).
update_guardrail ¶
update_guardrail(ws_id: WsRef, gid: Ref, name: str, detector_ids: Sequence[Ref], description: str = '', *, deny_suspicious_characters: bool = False) -> Guardrail
delete_guardrail ¶
Delete a workspace guardrail (ws_id) or an org guardrail.
test_guardrail ¶
test_guardrail(text: str, detector_ids: Sequence[Ref] | None = None, *, guardrail_id: Ref | None = None, deny_suspicious_characters: bool = False, ws_id: WsRef | None = None) -> GuardrailTest
Scan text through a guardrail without an LLM. Pass detector_ids, or
guardrail_id to reuse a saved guardrail's detectors and its
denySuspiciousCharacters setting. ws_id uses the workspace path, which
a workspace member can call. Returns finalDisposition plus a
results row per detector.
Builds without the guardrail tester (it answers 404/405) are scanned one
CUSTOM detector at a time with the saved-detector test. Those builds
cannot test a built-in (Governance) detector offline: its row has
status: "untested" and only a real request through the gateway
(chat_completion) shows what it does. finalDisposition then
covers the custom detectors only.
ensure_guardrail ¶
ensure_guardrail(ws_id: WsRef, name: str, detector_ids: Sequence[Ref], description: str = '', *, deny_suspicious_characters: bool = False, approve: bool = True) -> Ensured
Create or update the workspace guardrail name to use exactly these
detectors (ids or names); see ensure_detector.
policies ¶
A workspace's policies, or (no ws_id) the org's.
policy ¶
One policy, by the workspace path (ws_id, which also scopes a name
lookup) or the org-admin path.
create_policy ¶
create_policy(ws_id: WsRef, name: str, input_guardrail_ids: Sequence[Ref], output_guardrail_ids: Sequence[Ref], *, provider_ids: list[str] | None = None, description: str = '', response_scan_mode: str = 'post_stream') -> Policy
Guardrails may be given by name. provider_ids grants every model
of those providers. Some 8.0.2 builds refuse a policy that grants no
model (HTTP 422 "a policy must grant at least one model").
update_policy ¶
update_policy(ws_id: WsRef, pid: Ref, name: str, input_guardrail_ids: Sequence[Ref], output_guardrail_ids: Sequence[Ref], *, provider_ids: list[str] | None = None, description: str = '', response_scan_mode: str = 'post_stream') -> Policy
Replace a workspace policy. Made by an org admin, an edit to an ACTIVE
policy applies at once (verified on 8.0.2); otherwise check
hasPendingChange for an edit held for review.
delete_policy ¶
Delete a workspace policy (ws_id) or an org policy.
ensure_policy ¶
ensure_policy(ws_id: WsRef, name: str, input_guardrail_ids: Sequence[Ref], output_guardrail_ids: Sequence[Ref], *, provider_ids: list[str] | None = None, description: str = '', response_scan_mode: str = 'post_stream', approve: bool = True) -> Ensured
Create or update the workspace policy name; see ensure_detector.
submit ¶
kind: detectors | guardrails | policies. The object may be given by name.
withdraw ¶
Cancel a pending submission or pending change (kind as for submit).
approve ¶
Approve a submitted object, by id, record, or (with ws_id) name.
reject ¶
reject(kind: str, obj_id: Ref, reason: str = 'rejected', ws_id: WsRef | None = None) -> Detector | Guardrail | Policy
virtual_keys ¶
virtual_keys(ws_id: WsRef | None = None, *, status: str | None = None) -> list[VirtualKey]
A workspace's keys, or (no ws_id) every key; status filters.
virtual_key ¶
virtual_key(key_id: Ref, ws_id: WsRef | None = None) -> VirtualKey
One key, by the workspace path (ws_id) or the org-admin path. The
key may be given by id, name, record or its faig_... secret.
create_virtual_key ¶
create_virtual_key(ws_id: WsRef, name: str, *, policy_ids: Sequence[Ref] | None = None, provider_ids: list[str] | None = None, expires_at: str | None = None, tpm: int | str | None = None, rpm: int | str | None = None) -> NewVirtualKey
Create a key in a workspace; policies may be given by name. The
secret is in the response ONCE (secret_from_response) and never
readable again. Key names need not be unique.
update_virtual_key ¶
update_virtual_key(key_id: Ref, *, expires_at: str | None = None, tpm: int | str | None = None, rpm: int | str | None = None, policy_ids: Sequence[Ref] | None = None, provider_ids: list[str] | None = None, ws_id: WsRef | None = None) -> VirtualKey
The update REPLACES the record, so the current one is read and its
writable fields resent with the changes (8.0.2 rejects read-only fields
such as status or spend). policy_ids replaces the attached
policies (org-assigned policies stay forced onto the key regardless);
provider_ids replaces the providers granted.
rotate_virtual_key ¶
rotate_virtual_key(key_id: Ref, ws_id: WsRef | None = None) -> NewVirtualKey
New secret, once, in the response (secret_from_response).
revoke_virtual_key ¶
revoke_virtual_key(key_id: Ref, reason: str = 'revoked', ws_id: WsRef | None = None) -> VirtualKey
The secret stops working; 8.0.2 requires a reason.
renew_virtual_key ¶
renew_virtual_key(key_id: Ref, *, expires_in_days: float = 30, ws_id: WsRef | None = None) -> Renewed
Give a key a new secret, keeping its id, name, limits and policies.
8.0.2 rotates only an ACTIVE key ("a expired key cannot be rotated: mint
a new key instead"). An expired key becomes active again when its
expiresAt moves into the future, so an expired key is extended
expires_in_days from now and then rotated. A revoked key was shut off
on purpose and a suspended one has no API to lift it: both are refused
and nothing is changed. The old secret stops working at once.
ensure_virtual_key ¶
ensure_virtual_key(ws_id: WsRef, name: str, *, policy_ids: Sequence[Ref] | None = None, provider_ids: list[str] | None = None, expires_at: str | None = None, tpm: int | str | None = None, rpm: int | str | None = None) -> Ensured
Create the workspace's active key name, or bring an existing one's
policies, providers and limits in line (only the arguments given are
checked). A created key's secret is in Ensured.secret; an existing
key's secret cannot be read again (rotate it if you need one). Revoked
keys are ignored; two active keys with the name are refused.
model_groups ¶
model_groups(ws_id: WsRef | None = None) -> list[ModelGroup]
Model groups: the org's, or those a workspace can use.
workspace_models ¶
workspace_models(ws_id: WsRef) -> list[str]
The models a workspace allows (its allowedModels).
grantable_providers ¶
grantable_providers(ws_id: WsRef | None = None) -> list[GrantableProvider]
Providers, models and upstream keys that can be granted: to keys in a
workspace (ws_id), or org-wide.
attachable_policies ¶
Policies a key in this workspace can attach.
audit_logs ¶
audit_logs(offset: int = 0, limit: int = 25) -> AuditLogPage
One page of the org audit log, newest first. iter_audit_logs
filters by time, user, role, action and workspace, and follows pages.
audit_logs_page ¶
audit_logs_page(cursor: str | None = None, *, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, limit: int = LOG_PAGE, ws_id: WsRef | None = None) -> AuditLogPage
One page of audit log rows, cursor-paged like traffic_logs_page.
filters fields: user, role, action, message, workspace (spell
actions as audit_log_filter_options lists them: Approve, ...).
iter_audit_logs ¶
iter_audit_logs(*, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, max_items: int | None = None, page_size: int = LOG_PAGE, ws_id: WsRef | None = None) -> Iterator[AuditLogRow]
Every matching audit log row, newest first, following nextCursor.
audit_log_filter_options ¶
audit_log_filter_options(*, range: str | None = None, start: When | None = None, end: When | None = None, ws_id: WsRef | None = None) -> AuditFilterOptions
The values each audit filter can take (users, roles, actions, ...).
usage_summary ¶
usage_summary(frm: Day | None = None, to: Day | None = None, ws_id: WsRef | None = None) -> UsageSummary
Requests and tokens in the window, the previous window, and key/user
adoption. Dates are date objects or "YYYY-MM-DD"; the default
window is the last 30 days. ws_id scopes it to a workspace.
usage_trend ¶
usage_trend(frm: Day | None = None, to: Day | None = None, interval: str = 'daily', ws_id: WsRef | None = None) -> UsageTrend
{"interval", "points": [{"at", "requests", "tokens"}]}.
usage_rankings ¶
usage_rankings(frm: Day | None = None, to: Day | None = None, by: str = 'key', limit: int = 5, *, end: str | None = None, order_by: str | None = None, interval: str | None = None, ws_id: WsRef | None = None) -> Rankings
Top (or end="bottom") keys, users, workspaces or models by usage:
{"items": [...], "meta": {...}}. order_by: tokens | requests.
usage_breakdown ¶
usage_breakdown(by: str, item_id: str, frm: Day | None = None, to: Day | None = None, ws_id: WsRef | None = None) -> Breakdown
One ranked item broken down: a key, user or workspace by model, a
model by who used it. Read the rows with breakdown_items().
Rankings still list deleted keys (deleted: true); their breakdown
is a 404.
cost_summary ¶
cost_summary(frm: Day | None = None, to: Day | None = None, ws_id: WsRef | None = None) -> CostSummary
Spend (totals.chargebackSpend), volume and unit costs in the window.
cost_trend ¶
cost_trend(frm: Day | None = None, to: Day | None = None, interval: str = 'daily', ws_id: WsRef | None = None) -> CostTrend
cost_rankings ¶
cost_rankings(frm: Day | None = None, to: Day | None = None, by: str = 'key', limit: int = 5, *, end: str | None = None, order_by: str | None = None, interval: str | None = None, ws_id: WsRef | None = None) -> Rankings
Top spenders by key, user, workspace or model. order_by:
providerSpend | chargebackSpend | per1kTokens | perRequest.
cost_breakdown ¶
cost_breakdown(by: str, item_id: str, frm: Day | None = None, to: Day | None = None, ws_id: WsRef | None = None) -> Breakdown
analytics ¶
analytics(frm: Day | None = None, to: Day | None = None, limit: int = 10, ws_id: WsRef | None = None) -> Analytics
Guardrail analytics: violations by detector, key, user and workspace, severity, trend and scanner latency.
traffic_logs_page ¶
traffic_logs_page(cursor: str | None = None, *, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, limit: int = LOG_PAGE, ws_id: WsRef | None = None) -> TrafficLogPage
One page: {"items", "nextCursor", "hasNext", "total"} (total
on the first page only). Pass the previous nextCursor to go on.
traffic_logs ¶
traffic_logs(*, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, max_items: int | None = None, page_size: int = LOG_PAGE, ws_id: WsRef | None = None) -> Iterator[TrafficLogRow]
Every matching traffic log row, newest first, following nextCursor.
range: 24h | 7d | month | all, or pass start/end
(datetime or epoch milliseconds) for a custom window. filters
maps a field (verdict, user, workspace, virtual_key, model,
traffic_type, request_path) to a value, a list of values (is), or
{"operator": ..., "value": [...]}. virtual_key takes the key's
NAME. Rows are summaries; the full record (tokens, cost, prompt,
detector) is traffic_log(uuid).
wait_for_traffic ¶
wait_for_traffic(since: When, *, filters: Mapping[str, Any] | None = None, count: int = 1, timeout: float = 60, interval: float = 2, ws_id: WsRef | None = None) -> list[TrafficLogRow]
Poll until count traffic log rows matching filters exist from
since (a datetime or epoch ms, taken BEFORE sending the request)
on; rows land a few seconds after the request. Returns them, newest
first; raises TimeoutError after timeout seconds.
traffic_log ¶
traffic_log(uuid: str, ws_id: WsRef | None = None) -> TrafficLog
The full record: model, verdict, action, detector, guardrail, policy, prompt, cost, latency, http status, request id.
blocked_requests ¶
blocked_requests(*, since: When | None = None, until: When | None = None, range: str = '24h', max_items: int | None = None, ws_id: WsRef | None = None) -> list[BlockedRequest]
Requests the gateway blocked, newest first, each with what was sent,
which detectors fired and why, and the virtual key that sent it (id,
status, owner name and email). A traffic log names its key only by
keyAlias and prefix, so the key is looked up in the same scope.
since/until (datetime or epoch ms; until defaults to
now) give a custom window, for a poller's last-pull time; otherwise
range. One traffic_log call per row: bound it with max_items.
traffic_log_filter_options ¶
traffic_log_filter_options(*, range: str = '24h', start: When | None = None, end: When | None = None, ws_id: WsRef | None = None) -> TrafficFilterOptions
The values each filter can take in the window (models, users, keys, ...).
icap_logs_page ¶
icap_logs_page(offset: int = 0, *, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, limit: int = LOG_PAGE, ws_id: WsRef | None = None) -> IcapLogPage
One page of ICAP logs (FortiGate-inspected traffic; offset paging). 404 when the appliance has ICAP turned off.
icap_logs ¶
icap_logs(*, range: str = '24h', start: When | None = None, end: When | None = None, filters: Mapping[str, Any] | None = None, max_items: int | None = None, page_size: int = LOG_PAGE, ws_id: WsRef | None = None) -> Iterator[dict[str, Any]]
Every matching ICAP log row. filters fields: verdict, direction,
user, application, fortigate, shadow_ai, inspection.
icap_log ¶
icap_log(uuid: str, ws_id: WsRef | None = None) -> dict[str, Any]
One ICAP log record. Not verified live (ICAP was off on the test appliance), so it is untyped.
icap_log_filter_options ¶
icap_log_filter_options(*, range: str = '24h', start: When | None = None, end: When | None = None, ws_id: WsRef | None = None) -> dict[str, Any]
check_key ¶
check_key(api_key: str) -> KeyCheck
Is a virtual key secret accepted by the gateway? Returns valid and
model_count; only an unexpected status raises.
available_models ¶
The model names a virtual key may call (what chat_completion
takes as model). A model missing here fails with 404 "model not
available to this API key".
chat_completion ¶
chat_completion(api_key: str, model: str, messages: list[dict[str, Any]], *, max_tokens: int | None = None, temperature: float | None = None, timeout: float = 120, **extra: Any) -> ChatCompletion
One OpenAI-compatible chat completion through the gateway; the key's
policies apply. A guardrail block is HTTP 200 with finish_reason
content_filter (see blocked), not an exception.
chat_completion_stream ¶
chat_completion_stream(api_key: str, model: str, messages: list[dict[str, Any]], *, max_tokens: int | None = None, temperature: float | None = None, timeout: float = 120, **extra: Any) -> Iterator[dict[str, Any]]
A streamed chat completion: yields each chat.completion.chunk as
it arrives. A guardrail block arrives as chunks too (the block message,
then finish_reason: content_filter), including one raised by
in_stream response scanning mid-answer. collect_stream joins
the chunks into one completion that blocked understands.
Runs on its own connection and is never retried (it is a POST).