pyfortisiem

The first Python client for the FortiSIEM REST API — async-first, fully typed, and the only library that speaks the FortiSIEM 8.0 MCP server.

The pyforti* family covers FortiGate, FortiManager, and FortiZTP — but nothing for FortiSIEM. pyfortisiem fills that gap with a pooled async incident/event client, a sync client that automates all three FortiSIEM auth flows (including the CSRF-gated credential-management surface), and a typed session over the new MCP tool catalogue.

Not affiliated with, authorized, maintained, sponsored, or endorsed by Fortinet. See NOTICE.

Install

pip install pyfortisiem

Quickstart — async incidents + events

FortiSIEM event retrieval is server-side asynchronous (submit query → poll progress → page result). pyfortisiem maps that onto asyncio so many incidents enrich concurrently under one shared concurrency gate.

import asyncio
from pyfortisiem import AsyncFortiSIEMClient

async def main():
    async with AsyncFortiSIEMClient.with_basic_auth(
        "fortisiem.example.com:443", "admin", "pw", domain="Super",
        max_concurrency=10,
    ) as fsm:
        # marquee: one incident + its triggering events, fetched concurrently
        inc = await fsm.get_incident_with_events(123456)
        print(inc.incident_title, len(inc.events))

        # bulk: list incidents and enrich each with events, bounded by the gate
        incs = await fsm.list_incidents_with_events(status=[0], size=50)

asyncio.run(main())

Auth modes

The h5 surface is CSRF-protected: every state-changing call must echo the s session cookie back hex-encoded as an s: header (the GUI's hexEncode(getCookie("s"))). The sync client handles this for you:

from pyfortisiem import FortiSIEMClient

c = FortiSIEMClient("fortisiem.example.com:443", "admin", "pw", domain="Super")
cred = c.create_oauth_credential("automation")        # h5 / CSRF flow
token = c.mint_token(cred.client_id, cred.client_secret)
print([t.name for t in c.mcp_list_tools(token)])      # OAuth bearer → MCP

MCP — FortiSIEM 8.0

MCPSession wraps all 17 tools the FortiSIEM MCP server advertises, each with a typed pydantic input model and decoded response. The server double-wraps payloads in nested content[].text envelopes (sometimes JSON, sometimes Python repr); unwrap_mcp_content peels that off and the response models type what's underneath.

from pyfortisiem import FortiSIEMClient
from pyfortisiem.models import EntityIncidentQuery

c = FortiSIEMClient("fortisiem.example.com:443", "admin", "pw")
token = c.mint_token(client_id, client_secret)
with c.mcp_session(token) as s:
    for inc in s.get_incidents_by_entity(EntityIncidentQuery(ip="203.0.113.7")):
        print(inc.incident_id, inc.incident_title)

Two incident shapes

The MCP entity tools (get_incidents_by_entity, get_related_incidents_by_id*) return curated snake_case IncidentSummary rows, while get_incident_by_id's .data and the REST surface return the raw camelCase Incident. The shared models — Incident, Event, EventTable, MutationAck — are reused across both surfaces so you learn them once.

Concurrency & query lifecycle

max_concurrency (default 10) caps in-flight requests via one shared semaphore, and the httpx connection pool is sized to match. For large evidence sets, pass use_query_lifecycle=True to drive the triggeringEvents/start → progress → result async path instead of the direct endpoint.

Development

pip install -e ".[test]"
pytest                 # offline suites — no appliance needed
pytest -m live         # opt-in live tests; needs FSM_* env (see examples/)

License

MIT — see LICENSE.