Open Source · Apache 2.0 License
OrgKernel.
Trust for any agent.
Three primitives every enterprise agent needs: verifiable identity, bounded execution scope, and tamper-proof audit. Works with LangGraph, CrewAI, AutoGen, or any framework. No AURA dependency required.
Why OrgKernel
Every enterprise that deploys AI agents asks four questions: Who is this agent? What can it do? What did it do? Who authorized it? No agent framework answers these. OrgKernel does — in a few lines of code.
OrgKernel is a complete, production-grade solution for enterprise agent trust. It signs identities, enforces scopes, and audits every action. Your platform provides the execution runtime.
Installation
Requires Python 3.10+.
From source
git clone https://github.com/metaprise/orgkernel.git
cd orgkernel
pip install -e ".[postgres]" # postgres | mysql | sqlite
Database drivers
pip install orgkernel[postgres] # PostgreSQL (recommended)
pip install orgkernel[mysql] # MySQL / MariaDB
pip install orgkernel[sqlite] # SQLite for local dev
Quick start
Service layer
Direct Python async — no HTTP, no API key.
from datetime import datetime, timedelta, timezone
from sqlalchemy.ext.asyncio import AsyncSession
from orgkernel.database import async_engine, init_db, get_session_factory
from orgkernel.services import AgentIdentityService, ExecutionTokenService, AuditChainService
from orgkernel.schemas import AgentIdentityCSR, ExecutionTokenCreate, ScopeCheckRequest, AuditLayer
async def demo(db: AsyncSession) -> None:
identity_svc = AgentIdentityService(db)
token_svc = ExecutionTokenService(db)
audit_svc = AuditChainService()
# 1. Issue agent identity
csr = AgentIdentityCSR(
agent_name="invoice-processor",
org_id="acme-corp",
requested_ou="finance_team",
public_key="<agent-ed25519-public-key-base64url>",
purpose="automated-invoice-processing",
)
issued = await identity_svc.issue_from_csr(csr)
print(f"Agent ID: {issued.identity.agent_id}")
# 2. Mint scoped execution token
token = await token_svc.mint(
ExecutionTokenCreate(
agent_id=issued.identity.agent_id,
mission_id="msn_invoice01",
execution_scope=["read_invoice", "write_payment_draft"],
expires_at=datetime.now(timezone.utc) + timedelta(hours=4),
)
)
print(f"Token: {token.token_id}")
# 3. Enforce scope before every tool call
allowed = await token_svc.check_scope(
ScopeCheckRequest(
token_id=token.token_id,
tool_name="read_invoice",
params={"invoice_id": "4521"},
)
)
print(f"Scope check: passed={allowed.passed}")
# 4. Audit every action
chain_id = await audit_svc.initialize(
db,
mission_id="msn_invoice01",
agent_id=issued.identity.agent_id,
)
await audit_svc.append(
db,
chain_id=chain_id,
layer=AuditLayer.EXECUTION,
event="EXECUTION_tool_call",
agent_id=issued.identity.agent_id,
mission_id="msn_invoice01",
data={
"tool": "read_invoice",
"invoice_id": "4521",
"result": "success",
},
token_id=token.token_id,
)
# 5. Verify audit integrity
assert await audit_svc.verify_integrity(
db,
chain_id,
) is True
await db.commit()
FastAPI router
Expose the same primitives as an HTTP API.
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from orgkernel.database import async_engine, get_session_factory, init_db
from orgkernel.pyapi.router import router
app = FastAPI(title="Agent Platform powered by OrgKernel")
@app.on_event("startup")
async def startup():
async_engine.url = "postgresql+asyncpg://user:pass@localhost:5432/orgkernel"
await init_db()
async def get_db():
factory = get_session_factory(async_engine)
async with factory() as session:
yield session
app.include_router(
router,
prefix="/orgkernel",
get_db=get_db,
)
27 REST endpoints across identity, token, and audit — no API key required.
AgentIdentity
Cryptographically signed organizational credential. Ed25519 keypair with Org CA certificate chain. Revocable, time-bounded, tied to an organizational unit.
Issue from CSR
from orgkernel.schemas import AgentIdentityCSR
identity_svc = AgentIdentityService(db)
csr = AgentIdentityCSR(
agent_name="invoice-processor",
org_id="acme-corp",
requested_ou="finance_team",
public_key="<agent-ed25519-public-key-base64url>",
purpose="automated-invoice-processing",
)
result = await identity_svc.issue_from_csr(csr)
# result.agent_private_key_pem — returned once, never stored server-side
Static verification
Status + expiry check.
from orgkernel.schemas import AgentIdentityVerify
result = await identity_svc.verify(
AgentIdentityVerify(agent_id="aid_7f3k9..."),
)
print(result.is_active) # True / False
Challenge-response
Cryptographic proof of key possession.
from orgkernel.schemas import ChallengeResponse
challenge = await identity_svc.request_challenge(
agent_id="aid_7f3k9...",
issued_by="platform-gateway",
ttl_seconds=300,
)
# Agent signs challenge.nonce with private key, then:
result = await identity_svc.verify_challenge(
ChallengeResponse(
challenge_id=challenge.challenge_id,
agent_id="aid_7f3k9...",
certificate_id="aid_7f3k9...",
public_key="<agent-public-key-base64url>",
nonce=nonce,
signature="<agent-ed25519-signature>",
)
)
assert result.overall_valid is True
Identity lifecycle
from orgkernel.schemas import AgentIdentityRevoke
await identity_svc.suspend(agent_id="aid_7f3k9...") # → SUSPENDED
await identity_svc.reactivate(agent_id="aid_7f3k9...") # → ACTIVE
await identity_svc.revoke(
agent_id="aid_7f3k9...",
data=AgentIdentityRevoke(
revoked_by="aid_admin01",
reason="compromised",
),
) # → REVOKED (permanent)
ExecutionToken
Scoped, time-bounded permission token. Every tool call is checked against the token — out-of-scope calls are blocked before reaching any external system.
Mint
from datetime import datetime, timedelta, timezone
from orgkernel.schemas import ExecutionTokenCreate
token_svc = ExecutionTokenService(db)
token = await token_svc.mint(
ExecutionTokenCreate(
agent_id="aid_7f3k9...",
mission_id="msn_invoice01",
execution_scope=["read_invoice", "write_payment_draft"],
immutable_params={"currency": "USD"},
bounded_params=[
{"name": "amount", "upper_bound": 50000},
],
expires_at=datetime.now(timezone.utc) + timedelta(hours=4),
)
)
print(token.token_id) # → "tok_abc123..."
print(token.token_signature) # Org CA signature (prevents Token Grafting)
Enforce scope
Never raises, returns a result.
from orgkernel.schemas import ScopeCheckRequest
result = await token_svc.check_scope(
ScopeCheckRequest(
token_id=token.token_id,
tool_name="read_invoice",
params={"invoice_id": "4521"},
)
)
print(result.passed) # → True / False
print(result.blocked) # → True if scope violation
Token lifecycle
await token_svc.mark_used(
token_id="tok_abc123...",
) # consumed
await token_svc.invalidate(
token_id="tok_abc123...",
reason="mission aborted",
) # early invalidation
AuditChain
Append-only, hash-chained execution log. Each entry is SHA-256 linked to the previous. Audit is never optional — it is synchronous.
Initialize and append
from orgkernel.schemas import AuditLayer
audit_svc = AuditChainService()
chain_id = await audit_svc.initialize(
db,
mission_id="msn_invoice01",
agent_id="aid_7f3k9...",
)
# Writes genesis IDENTITY entry automatically
await audit_svc.append(
db,
chain_id=chain_id,
layer=AuditLayer.EXECUTION,
event="EXECUTION_tool_call",
agent_id="aid_7f3k9...",
mission_id="msn_invoice01",
data={
"tool": "accounting_api",
"invoice_id": "4521",
"result": "success",
"duration_ms": 230,
},
token_id="tok_abc123...",
)
await audit_svc.close(db, chain_id=chain_id)
Verify integrity
Sequence continuity, hash chain validity, entry self-consistency.
valid = await audit_svc.verify_integrity(
db,
chain_id,
)
assert valid is True # → True / False
Retrieve the full chain
chain = await audit_svc.get_by_chain_id(db, chain_id)
for entry in chain.entries:
print(f" [{entry.layer}] {entry.event}: {entry.data}")
What OrgKernel is Not
OrgKernel is a trust library — it signs identities, enforces scopes, and audits actions. Your platform provides everything else.
The following are Metaprise hosted platform concepts — they do not exist in OrgKernel:
- Mission orchestration — OrgKernel has no execution runtime; you control mission scheduling
- Agent registry / managed execution — no
agents.deploy,agents.run, oragents.list - Mission marketplace — no registry of publishable or installable agents
- Streaming state transitions — OrgKernel records state, it does not drive it
License
Copyright 2026 Metaprise
Licensed under the Apache License, Version 2.0.