Metadata-Version: 2.4
Name: globiguard
Version: 0.3.0
Summary: Official dependency-minimal Python SDK for GlobiGuard.
Author: GlobiGuard
License-Expression: Apache-2.0
Project-URL: Homepage, https://globiguard.com
Project-URL: Repository, https://github.com/globiguard/globiguard-python
Project-URL: Issues, https://github.com/globiguard/globiguard-python/issues
Keywords: ai-governance,compliance,audit,approval,webhooks
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# globiguard-python

Official dependency-minimal Python SDK for GlobiGuard.

This package provides a small, auditable integration surface for production services. It intentionally uses the Python standard library for runtime behavior: no `requests`, no `httpx`, no pydantic, no framework dependency, and no hidden telemetry.

## What is included

- Server and browser-style clients with the same auth headers as the TypeScript SDK.
- Resource clients for actions, audit, installs, orgs/API keys, policies, queue, and workflows.
- Governed action helpers for authorize-or-throw, approval polling, correlation IDs, and idempotency keys.
- AI intercept for 10 providers: OpenAI, Anthropic, Google GenAI, AWS Bedrock, Cohere, Mistral, Ollama, LangChain, LlamaIndex, and any callable via `generic()`.
- Multi-agent governance: `GovernanceContext` records input/tool-call/output hops under a shared `correlation_id` for LangGraph, CrewAI, AutoGPT, n8n, and generic orchestration frameworks.
- Inter-agent trust tokens: HMAC-signed `AgentToken` issued per agent, verified by downstream agents, with automatic BLOCK propagation.
- Bootstrap install registration/heartbeat builders for hosted, self-hosted, and sovereign deployment modes.
- Offline entitlement manifest verification for customer-controlled deployments.
- Server-only trust webhook verification using raw request bodies and HMAC-SHA256.
- Typed package marker (`py.typed`) and a small, inspectable source layout.

## Install

```bash
pip install globiguard
```

To test the unreleased local `sol` branch instead of the latest PyPI release,
install from the repository root:

```bash
python -m pip install .
```

## Authentication and keys

GlobiGuard project IDs, secret keys, publishable keys, local credentials, and webhook signing secrets are issued by the GlobiGuard app/control plane. The Python SDK uses the same wire headers as the TypeScript SDK:

| Credential | Headers |
| --- | --- |
| Secret | `x-globiguard-project-id`, `x-globiguard-secret-key` |
| Publishable | `x-globiguard-project-id`, `x-globiguard-publishable-key` |
| Local | `x-globiguard-local-mode`, optional `x-globiguard-local-token` |
| All clients | `x-globiguard-client`, `x-globiguard-environment` |

Server clients only accept secret or local credentials. Publishable keys are limited to read-only/browser-style clients. Local credentials only work with `environment="local"` and localhost/loopback service URLs.

## Server client

```python
import globiguard

client = globiguard.create_server_client(
    environment="sandbox",
    services={"controlPlane": "https://api.globiguard.com"},
    credential=globiguard.SecretCredential(
        project_id="proj_123",
        token="sk_live_...",
        environment="sandbox",
    ),
)

decision = client.governed_actions.authorize_action_or_throw(
    {
        "context": {
            "actionType": "refund.create",
            "destination": {
                "type": "custom",
                "name": "payments-production",
            },
            "dataClasses": ["CONFIDENTIAL"],
            "actor": {
                "id": "support-agent-123",
                "type": "agent",
            },
            "purpose": "Resolve an approved customer escalation",
            "correlationId": "case_456",
            "idempotencyKey": "case_456:refund:v1",
        }
    }
)
```

The helper returns only a current, short-lived, obligation-free `ALLOW` that
explicitly authorizes the exact action once. `MODIFY`, `QUEUE`, `BLOCK`,
dry-run, expired, and incomplete responses raise `GlobiguardAuthorityError`
and keep the downstream action stopped.

Optional action gateway routing matches the TypeScript SDK:

```python
client = globiguard.create_server_client(
    environment="local",
    services={
        "controlPlane": "http://localhost:3000",
        "sidecar": "http://localhost:8787",
    },
    credential=globiguard.LocalCredential(token="dev-token"),
    action_gateway=globiguard.ActionGatewayConfig(mode="sidecar"),
)
```

## Resource clients

```python
client.actions.authorize({...})
client.actions.get_authorization("auth_123")
client.actions.create_approval({...})
client.actions.list_evidence(authorization_id="auth_123")

client.audit.list(from_="2026-05-01T00:00:00Z", limit=50)
client.audit.export({"format": "json"})
client.audit.get_evidence_package_summary("pkg_123")
client.audit.get_incident_replay(correlation_id="corr_123")

client.installs.register({...})
client.installs.heartbeat("install_123", {...})

client.orgs.find_by_slug("acme")
client.orgs.create_api_key("org_123", {...})
client.orgs.revoke_api_key("org_123", "key_123")

client.policies.list(active=True)
client.policies.create_from_template("tpl_123")
client.policies.activate("pol_123")

client.queue.list(status="PENDING")
client.queue.decide("queue_123", action="approve", reviewed_by="user_123")
client.queue.decide(
    "queue_456",
    action="modify",
    reviewed_by="user_123",
    reason_code="REMOVE_SSN",
    modified_payload_summary={
        "sha256": "reviewed-payload-digest",
        "fieldTypes": ["CUSTOMER_ID"],
    },
)

client.workflows.list(active=True)
client.workflows.run("wf_123", {"source": "python"})
client.workflows.list_runs("wf_123")
```

## AI intercept

`AiIntercept` wraps any AI provider call with a GlobiGuard governance checkpoint. Input is authorized before the model is called; output is classified and authorized if sensitive. Supports OpenAI, Anthropic, Google GenAI, AWS Bedrock, Cohere, Mistral, Ollama, LangChain, LlamaIndex, and any callable via `generic()`.

```python
from globiguard import create_server_client, SecretCredential
from globiguard.ai_intercept import AiIntercept
import openai

client = create_server_client(
    environment="live",
    services={"controlPlane": "https://api.globiguard.com", "brain": "https://brain.globiguard.com"},
    credential=SecretCredential(project_id="proj_123", token="sk_...", environment="live"),
)

intercept = AiIntercept(client.governed_actions, brain=client.brain)

# OpenAI — returns a proxy, call exactly like the original client
governed_openai = intercept.openai(openai.OpenAI())
response = governed_openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarise this contract..."}],
)

# Anthropic
import anthropic
governed_anthropic = intercept.anthropic(anthropic.Anthropic())
msg = governed_anthropic.messages.create(model="claude-opus-4-8", max_tokens=1024, messages=[...])

# Google GenAI
import google.generativeai as genai
governed_model = intercept.google(genai.GenerativeModel("gemini-1.5-pro"))
result = governed_model.generate_content("Draft a privacy policy...")

# AWS Bedrock
import boto3
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
governed_bedrock = intercept.bedrock(bedrock)
response = governed_bedrock.converse(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[{"role": "user", "content": [{"text": "Hello"}]}],
)

# Cohere
import cohere
governed_cohere = intercept.cohere(cohere.Client("api-key"))
response = governed_cohere.chat(message="Summarise...")

# Mistral
from mistralai import Mistral
governed_mistral = intercept.mistral(Mistral(api_key="..."))
response = governed_mistral.chat.complete(model="mistral-large-latest", messages=[...])

# Ollama
import ollama
governed_ollama = intercept.ollama(ollama.Client())
response = governed_ollama.chat(model="llama3", messages=[{"role": "user", "content": "Hello"}])

# LangChain — callback handler, works across all LangChain providers
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
handler = intercept.langchain_callback()
result = llm.invoke("Draft a contract...", config={"callbacks": [handler]})

# LlamaIndex — callback manager
from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
Settings.callback_manager = CallbackManager([intercept.llamaindex_callback()])

# Any provider via generic()
def my_provider(prompt: str) -> str: ...
governed_fn = intercept.generic(my_provider, extract_input=lambda p: p)
result = governed_fn(prompt="Hello")
```

By default, `AiIntercept` runs in `scan_both` mode — it scans input before the call and output after. Set `mode="scan_input"` or `mode="scan_output"` to restrict scope. When a governance decision is `BLOCK`, `GlobiguardAuthorityError` is raised; pass `on_block` to handle it yourself instead.

## Multi-agent governance

`GovernanceContext` wraps a single agent step — it records input scan, tool-call, and output scan results as typed hops under a shared `correlation_id`. All hops for the same workflow are stitched into one trace visible in the GlobiGuard portal.

```python
from globiguard import create_server_client, SecretCredential
from globiguard.governance import GovernanceContext

client = create_server_client(
    environment="live",
    services={"controlPlane": "https://api.globiguard.com", "brain": "https://brain.globiguard.com"},
    credential=SecretCredential(project_id="proj_123", token="sk_...", environment="live"),
)

# Works with LangGraph, CrewAI, AutoGPT, n8n, or any agentic framework.
with GovernanceContext(
    client.governance,
    org_id="org_123",
    session_id="sess_abc",
    correlation_id="corr_wf_run_001",   # shared across all agents in the workflow
    agent_id="classify_node",
    framework="langgraph",
    workflow_name="patient_intake",
) as ctx:
    # Scan input text before passing to LLM
    input_eval = client.brain.evaluate(text=user_message, industry="HEALTHCARE", session_id="sess_abc", org_id="org_123")
    ctx.record_input(input_eval)

    # ... call LLM, produce output ...

    # Scan output before returning or writing to CRM
    output_eval = client.brain.evaluate(text=llm_output, industry="HEALTHCARE", session_id="sess_abc", org_id="org_123")
    ctx.record_output(output_eval)  # raises GovernanceContext.GovernanceBlockedError on BLOCK
```

Pass `raise_on_block=False` to handle BLOCK decisions yourself instead of throwing.

The low-level client is at `client.governance`:

```python
# Record a hop manually (useful when calling evaluate via context dict)
client.governance.record_hop(
    correlation_id="corr_wf_run_001",
    org_id="org_123",
    session_id="sess_abc",
    decision="ALLOW",
    phase="tool_call",
    tool_name="crm.write",
    framework="crewai",
    latency_ms=12.4,
)

# Fetch a full trace
trace = client.governance.get_trace("gtrace_abc123")

# List recent traces for an org
traces = client.governance.list_traces("org_123", limit=20)
```

### Inter-agent trust tokens

When agent B receives a call from agent A, B can verify A's identity before acting on the instruction — preventing prompt-injection attacks from impersonating trusted upstream agents.

```python
# Agent A — issue a token at the start of its execution
token = client.governance.issue_token(
    agent_id="extractor_agent",
    org_id="org_123",
    session_id="sess_abc",
    correlation_id="corr_wf_run_001",
    ttl_seconds=300,
)

# Agent B — verify the token before accepting A's output
result = client.governance.verify_token(
    token,
    expected_correlation_id="corr_wf_run_001",
)

if result["verdict"] != "trusted":
    raise RuntimeError(f"Upstream agent not trusted: {result['reason']}")
```

If the upstream agent received a `BLOCK` decision, `verify_token` automatically returns `verdict="untrusted"` with `reason="upstream_agent_blocked"` — no manual check needed.

## Bootstrap installs

Bootstrap helpers keep install registration and heartbeat payloads aligned with the hosted/self-hosted/sovereign deployment contract.

```python
profile = {
    "environment": "sandbox",
    "deploymentMode": "self_hosted",
    "issuerMode": "customer_issued",
    "installReporting": "opt_in",
    "installLabel": "Python worker",
}

registration = globiguard.build_install_registration_request(
    profile,
    package_name="globiguard",
    package_version="0.3.0",
    integration_kind="sdk",
    runtime_kind="python",
)
```

Hosted deployments must use `globiguard_issued`. Self-hosted and sovereign deployments must use `customer_issued` and explicitly choose `opt_in` or `disabled` install reporting.

## Offline entitlement manifests

Customer-controlled deployments can verify signed entitlement manifests without calling GlobiGuard at runtime.

```python
payload = globiguard.verify_signed_entitlement_manifest(
    manifest,
    public_keys_by_id={"kid_2026_05": "<base64url-ed25519-public-key>"},
    expected_issuer="https://api.globiguard.com",
    expected_org_id="org_123",
    expected_project_id="proj_123",
    expected_environment="live",
    expected_deployment_mode="self_hosted",
)
```

The verifier checks compact JWS structure, protected header/payload consistency, Ed25519 signature, manifest schema, timestamp validity, and optional issuer/workspace/project/environment/deployment expectations.

## Webhook verification

Pass the exact raw request body bytes received from your framework. Do not parse and re-serialize JSON before verification; whitespace, key order, and line endings are part of the signed payload.

```python
result = globiguard.verify_trust_webhook(
    headers=request.headers,
    raw_body=request.get_data(),
    signing_secret="whsec_...",
)

if not result["ok"]:
    raise ValueError(result["error"]["message"])

event = result["envelope"]
```

Webhook verification checks required headers, timestamp replay window, envelope/header consistency, and `v1=` HMAC signatures using constant-time comparison.

## Security posture

- Runtime dependencies: **zero**.
- HTTPS is required outside `local`.
- Local credentials require localhost or loopback URLs.
- Reserved GlobiGuard auth headers cannot be overridden by per-request headers.
- Credential `repr()` output redacts tokens.
- Request paths reject absolute URLs, query strings, fragments, backslashes, invalid percent encoding, and dot segments.
- HTTP requests use explicit timeouts.
- Entitlement manifests verify Ed25519 signatures locally without a crypto package dependency.

Known gap: the main app has two adjacent key surfaces: workspace API keys under `/app/api-keys` and SDK project credentials under `/projects/:projectId/credentials`. Python integrations need the SDK project credentials (`sandbox` or `live`) so the UI/docs should make that distinction very clear.

## Development

```bash
set PYTHONPATH=src
python -m unittest discover -s tests -v
python -m compileall -q src tests
python -m pip install --no-deps --no-build-isolation .
```
