aigp-client

aigp-client

Universal AI Governance Protocol (AIGP) client — RFC-010 implementation with provider proxy.

Install

pip install aigp-client>=3.0.0

Usage — Provider Proxy

The recommended way to use aigp-client is via the provider proxy methods. These wrap every AI call with governance (check → invoke → record) in a single call. No raw boto3 needed.

from aigp_client import AigpClient

client = AigpClient(
    gov_url="https://www.your-governance-server.com",
    app_id="MY_APP",
    hmac_secret="your-shared-secret",
    mode="REPORT",  # or "ENFORCE"
)

# Invoke a model — returns text response
text = await client.invoke_text(
    model_id="us.amazon.nova-pro-v1:0",
    prompt="Summarize this document",
    system_prompt="You are a helpful assistant.",
    use_case="summarization",
    user_id="user@example.com",
    region="us-east-1",
)

# Full Converse response (with usage metadata)
resp = await client.invoke(
    model_id="us.amazon.nova-pro-v1:0",
    messages=[{"role": "user", "content": [{"text": "Hello"}]}],
    system_prompt="You are helpful.",
    use_case="chat",
)

# Retrieve from a Bedrock Knowledge Base
results = await client.retrieve(
    kb_id="CN7UAQYKMG",
    query="What are the HIPAA requirements?",
    num_results=5,
    use_case="compliance",
)

# RAG — Retrieve & Generate
answer = await client.retrieve_and_generate(
    kb_id="CN7UAQYKMG",
    query="Assess our AI governance posture",
    model_arn="arn:aws:bedrock:us-east-1::foundation-model/us.amazon.nova-pro-v1:0",
    use_case="governance_assessment",
)

Each proxy method automatically:

1. CHECK — pre-invocation policy check with governance-server
2. INVOKE — calls Bedrock (Converse API or KB API)
3. RECORD — post-invocation telemetry (tokens, duration, status)

If governance denies the request in ENFORCE mode, a ValueError is raised.

Low-Level Protocol Methods

For custom integrations or non-Bedrock providers:

# Heartbeat (run as background task)
await client.heartbeat()

# Pre-invocation check
decision = await client.check("my_use_case", "model-id", user_id="user@example.com")
if decision.denied:
    raise Exception(f"Blocked: {decision.reason}")

# Post-invocation record
await client.record(
    use_case="my_use_case", model_id="model-id",
    input_tokens=500, output_tokens=200,
    duration_ms=1200, user_id="user@example.com",
)

Modes

Protocol (RFC-010)

All messages are HMAC-SHA256 signed with headers:

• X-AIGP-Signature: hmac-sha256={sig}
• X-AIGP-Timestamp: {iso_timestamp}
• X-AIGP-App-Id: {app_id}

Use Cases Config

Ship an aigp-use-cases.json alongside your app:

{
  "app_id": "MY_APP",
  "use_cases": [
    {"id": "chat", "description": "General AI chat"},
    {"id": "summarization", "description": "Document summarization"},
    {"id": "compliance", "description": "Compliance KB queries"}
  ]
}

Auto-discovered at ./aigp-use-cases.json or /app/aigp-use-cases.json, or pass explicitly:

client = AigpClient(..., use_cases_file="/path/to/aigp-use-cases.json")

Docker Integration

RUN pip install aigp-client>=3.0.0
COPY aigp-use-cases.json /app/aigp-use-cases.json

Migration from v1.0.0

Replace manual check→invoke→record patterns:

# Before (v1.0.0) — manual governance wrapper
decision = await client.check("chat", model_id)
if decision.denied:
    raise ...
response = bedrock.invoke_model(...)  # raw boto3
await client.record("chat", model_id, in_tok, out_tok, duration)

# After (v1.1.0) — single call, governance built-in
text = await client.invoke_text(model_id, prompt, use_case="chat")

Version History

• 1.1.0 — Provider proxy methods (invoke, invoke_text, retrieve, retrieve_and_generate). No more raw boto3 needed.
• 1.0.0 — Initial release. Low-level protocol methods (check, record, heartbeat).