|
| 1 | +"""Example: explain a previously-made AxonFlow policy decision. |
| 2 | +
|
| 3 | +Implements the ADR-043 explainability flow. Given a decision_id (typically |
| 4 | +surfaced on the response of a blocked governed call, an audit_logs row, or |
| 5 | +the ``explain_decision`` MCP tool), this example fetches the structured |
| 6 | +explanation and renders the matched policies, risk level, and override |
| 7 | +availability. |
| 8 | +
|
| 9 | +Required env vars: |
| 10 | +
|
| 11 | +* ``AXONFLOW_AGENT_URL`` (default: http://localhost:8080) |
| 12 | +* ``AXONFLOW_CLIENT_ID`` (default: community) |
| 13 | +* ``AXONFLOW_CLIENT_SECRET`` (default: empty) |
| 14 | +* ``AXONFLOW_DECISION_ID`` the decision to explain |
| 15 | +
|
| 16 | +Get a decision_id quickly by hitting a known-blocked policy:: |
| 17 | +
|
| 18 | + curl -u "$AXONFLOW_CLIENT_ID:$AXONFLOW_CLIENT_SECRET" \\ |
| 19 | + -X POST $AXONFLOW_AGENT_URL/api/v1/mcp/check-input \\ |
| 20 | + -H 'Content-Type: application/json' \\ |
| 21 | + -d '{"connector_type":"postgres","operation":"execute", |
| 22 | + "statement":"SELECT 1; DROP TABLE users;--","user_token":"u1"}' |
| 23 | +
|
| 24 | +then read decision_id from the block response or the most recent audit row. |
| 25 | +""" |
| 26 | + |
| 27 | +import asyncio |
| 28 | +import os |
| 29 | +import sys |
| 30 | + |
| 31 | +from axonflow import AxonFlow |
| 32 | + |
| 33 | + |
| 34 | +async def main() -> None: |
| 35 | + decision_id = os.environ.get("AXONFLOW_DECISION_ID", "") |
| 36 | + if not decision_id: |
| 37 | + print( |
| 38 | + "AXONFLOW_DECISION_ID must be set (a decision_id from a recent blocked call)", |
| 39 | + file=sys.stderr, |
| 40 | + ) |
| 41 | + sys.exit(2) |
| 42 | + |
| 43 | + endpoint = os.environ.get("AXONFLOW_AGENT_URL", "http://localhost:8080") |
| 44 | + print(f"Initializing AxonFlow client at {endpoint}...") |
| 45 | + async with AxonFlow( |
| 46 | + endpoint=endpoint, |
| 47 | + client_id=os.environ.get("AXONFLOW_CLIENT_ID", "community"), |
| 48 | + client_secret=os.environ.get("AXONFLOW_CLIENT_SECRET", ""), |
| 49 | + ) as client: |
| 50 | + print(f"Explaining decision {decision_id}...\n") |
| 51 | + exp = await client.explain_decision(decision_id) |
| 52 | + |
| 53 | + print("=== Decision Explanation ===") |
| 54 | + print(f" decision_id: {exp.decision_id}") |
| 55 | + print(f" timestamp: {exp.timestamp.isoformat()}") |
| 56 | + print(f" decision: {exp.decision}") |
| 57 | + print(f" reason: {exp.reason}") |
| 58 | + if exp.risk_level: |
| 59 | + print(f" risk_level: {exp.risk_level}") |
| 60 | + if exp.tool_signature: |
| 61 | + print(f" tool: {exp.tool_signature}") |
| 62 | + |
| 63 | + print(f"\n policy_matches ({len(exp.policy_matches)}):") |
| 64 | + for i, m in enumerate(exp.policy_matches): |
| 65 | + print( |
| 66 | + f" [{i}] {m.policy_id} ({m.policy_name or '(unnamed)'}) — " |
| 67 | + f"action={m.action or '-'} risk={m.risk_level or '-'} " |
| 68 | + f"allow_override={m.allow_override}" |
| 69 | + ) |
| 70 | + |
| 71 | + if exp.matched_rules: |
| 72 | + print(f"\n matched_rules ({len(exp.matched_rules)}):") |
| 73 | + for r in exp.matched_rules: |
| 74 | + print( |
| 75 | + f" {r.policy_id} on {r.rule_id or '(no rule id)'}: " |
| 76 | + f"matched={r.matched_on or '-'}" |
| 77 | + ) |
| 78 | + |
| 79 | + print(f"\n override_available: {exp.override_available}") |
| 80 | + if exp.override_existing_id: |
| 81 | + print(f" override_existing_id: {exp.override_existing_id}") |
| 82 | + print(f" historical_hit_count_session: {exp.historical_hit_count_session}") |
| 83 | + if exp.policy_source_link: |
| 84 | + print(f" policy_source_link: {exp.policy_source_link}") |
| 85 | + |
| 86 | + |
| 87 | +if __name__ == "__main__": |
| 88 | + asyncio.run(main()) |
0 commit comments