← All Cookbooks
OpenAIIntermediate20 min

OpenAI Agents + HatiData: Agent RBAC Setup

Set up per-agent API keys with scoped permissions in HatiData. Control what each OpenAI agent can read, write, and administer.

What You'll Build

A multi-agent system with per-agent API keys, scoped permissions, and automatic key rotation.

Prerequisites

$pip install openai hatidata-agent

$hati init

$OpenAI API key

Architecture

┌──────────────┐  API Key  ┌──────────────┐    ┌──────────────┐
│  Agent A     │─────────▶│  HatiData    │───▶│   Engine     │
│  (ReadOnly)  │          │  RBAC Engine │    │              │
└──────────────┘          └──────────────┘    └──────────────┘
┌──────────────┐  API Key  ┌──────────────┐
│  Agent B     │─────────▶│  HatiData    │
│  (Admin)     │          │  RBAC Engine │
└──────────────┘          └──────────────┘

Key Concepts

  • ●Per-agent API keys: each agent gets its own key with a specific scope, enabling fine-grained access control
  • ●Scope enforcement: readonly keys can only SELECT, admin keys have full access including DDL and DML
  • ●Key rotation: rotate_api_key() generates a new key and immediately invalidates the old one
  • ●Audit-ready: all key operations (creation, rotation, usage) are logged for security auditing

Step-by-Step Implementation

1

Install Dependencies

Install the OpenAI SDK and HatiData agent SDK.

Bash
pip install openai hatidata-agent
hati init
Expected Output
HatiData initialized successfully.
Proxy running on localhost:5439

Note: Each agent gets its own API key with specific scope. The proxy enforces permissions on every query.

2

Create Agent API Keys

Generate scoped API keys for different agent roles.

Python
from hatidata_agent import HatiDataAgent

admin = HatiDataAgent(host="localhost", port=5439, agent_id="admin")

# Create a read-only key for the reporting agent
result = admin.query("""
    SELECT create_api_key(
        'reporting-agent',
        'readonly',
        'API key for the reporting agent — read-only access'
    )
""")
readonly_key = result[0]['api_key']
print(f"Read-only key: {readonly_key[:12]}...")

# Create an admin key for the data management agent
result = admin.query("""
    SELECT create_api_key(
        'data-manager',
        'admin',
        'API key for the data management agent — full access'
    )
""")
admin_key = result[0]['api_key']
print(f"Admin key: {admin_key[:12]}...")
Expected Output
Read-only key: hd_ro_a1b2c3...
Admin key: hd_admin_d4e5...

Note: Key scopes: readonly (SELECT only), admin (full access including DDL). Keys are prefixed with their scope for easy identification.

3

Test Read-Only Access

Verify that the read-only agent cannot perform write operations.

Python
# Connect as the read-only reporting agent
reporter = HatiDataAgent(
    host="localhost", port=5439,
    agent_id="reporting-agent",
    api_key=readonly_key
)

# This works: SELECT query
result = reporter.query("SELECT COUNT(*) AS cnt FROM products")
print(f"SELECT query: OK — {result[0]['cnt']} rows")

# This is blocked: INSERT
try:
    reporter.execute("INSERT INTO products VALUES (99, 'test', 'test', 0, 0, 'test')")
    print("INSERT: OK")
except Exception as e:
    print(f"INSERT: BLOCKED — {e}")

# This is blocked: DROP
try:
    reporter.execute("DROP TABLE products")
    print("DROP: OK")
except Exception as e:
    print(f"DROP: BLOCKED — {e}")
Expected Output
SELECT query: OK — 42 rows
INSERT: BLOCKED — Permission denied: readonly scope cannot execute INSERT
DROP: BLOCKED — Permission denied: readonly scope cannot execute DDL
4

Test Admin Access

Verify that the admin agent has full access.

Python
# Connect as the admin data manager
manager = HatiDataAgent(
    host="localhost", port=5439,
    agent_id="data-manager",
    api_key=admin_key
)

# Admin can read
result = manager.query("SELECT COUNT(*) AS cnt FROM products")
print(f"SELECT: OK — {result[0]['cnt']} rows")

# Admin can write
manager.execute("INSERT INTO products VALUES (99, 'New Product', 'Test', 9.99, 5.0, 'A test product')")
print("INSERT: OK")

# Admin can create tables
manager.execute("CREATE TABLE IF NOT EXISTS audit_log (event VARCHAR, ts TIMESTAMP DEFAULT NOW())")
print("CREATE TABLE: OK")

# Clean up
manager.execute("DELETE FROM products WHERE id = 99")
manager.execute("DROP TABLE audit_log")
print("Cleanup: OK")
Expected Output
SELECT: OK — 42 rows
INSERT: OK
CREATE TABLE: OK
Cleanup: OK
5

Rotate Keys

Rotate an agent's API key and verify the old key is invalidated.

Python
# Rotate the reporting agent's key
result = admin.query("""
    SELECT rotate_api_key('reporting-agent')
""")
new_key = result[0]['api_key']
print(f"New key: {new_key[:12]}...")
print(f"Old key invalidated")

# Verify old key no longer works
try:
    old_reporter = HatiDataAgent(
        host="localhost", port=5439,
        agent_id="reporting-agent",
        api_key=readonly_key  # old key
    )
    old_reporter.query("SELECT 1")
    print("Old key: still works (unexpected)")
except Exception as e:
    print(f"Old key: rejected — {e}")

# List all active keys
keys = admin.query("""
    SELECT agent_id, scope, created_at, last_used_at
    FROM _hatidata_keys.active_keys
    ORDER BY created_at DESC
""")
print(f"\nActive API keys: {len(keys)}")
for k in keys:
    print(f"  {k['agent_id']}: scope={k['scope']}, created={k['created_at']}")
Expected Output
New key: hd_ro_f6g7h8...
Old key invalidated
Old key: rejected — Authentication failed: invalid API key

Active API keys: 2
  reporting-agent: scope=readonly, created=2026-02-28 15:00:05
  data-manager: scope=admin, created=2026-02-28 14:59:50

Note: Key rotation is instant. The old key is immediately invalidated, and the new key is active right away.

Related Use Case

Security

Cybersecurity Threat Hunting

Threat Hunting with Semantic Intelligence

Ready to build?

Install HatiData locally and start building with OpenAI in minutes.

Join Waitlist