The D2 Python SDK reduces the complexity of implementing enterprise-grade authorization for AI agents. Add authorization to any Python function with a single decorator.
pip install d2-pythonAfter installation, initialize D2 at application startup. Here are examples for different frameworks:
# main.py
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
import d2
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize D2 on startup
await d2.configure_rbac_async()
yield
# Cleanup on shutdown
await d2.shutdown_rbac()
app = FastAPI(lifespan=lifespan)
# Add D2 middleware for automatic context management
app.add_middleware(d2.ASGIMiddleware)
@app.get("/weather")
async def get_weather(location: str):
# Context automatically set by middleware from headers
return await weather_service.fetch(location)You can protect functions in two ways: with custom tool IDs or auto-detected IDs. The following examples show both approaches:
@d2.d2_guard("weather_api")
def get_weather(location: str):
return weather_service.fetch(location)Recommended: Use descriptive names for better policy organization.
@d2.d2_guard()
def send_email(recipient: str):
# Tool ID: myapp.send_email
return email_service.send(recipient)Zero config: Uses module.qualname format - the module path plus the qualified name (including nested classes - see below).
When you omit the first argument, D2 automatically generates a stable tool ID using module.qualname - the module path combined with the qualified name, which includes the full path through any nested classes or functions:
def health_check(): ...
class UserApi:
def get(self): ...myapp.health_check
myapp.UserApi.getD2 automatically scans your codebase and creates a policy template with all detected functions:
python -m d2 initThe d2 init command analyzes your Python files, finds all @d2_guard decorators, and automatically generates a policy template with the detected tool IDs.
metadata:
name: "my-app"
expires: "2025-02-01T00:00:00Z"
policies:
- role: admin
permissions: ["*"]
- role: developer
permissions:
- "weather_api" # Custom ID from your code
- "myapp.send_email" # Auto-detected IDThe D2 SDK provides flexible error handling options. You can customize how authorization denials are handled using the on_deny parameter:
@d2.d2_guard("admin_only")
def delete_user(user_id: str):
return user_service.delete(user_id)
# Handle the exception
try:
delete_user("user123")
except d2.PermissionDeniedError as e:
print(f"Access denied: {e}")
return {"error": "Insufficient permissions"}Default behavior: Raises PermissionDeniedError when access is denied.
The D2 SDK raises specific exceptions for different failure modes. All exceptions inherit from D2Error:
| Exception | When Raised |
|---|---|
| PermissionDeniedError | User lacks required permissions for the protected function |
| MissingPolicyError | No policy bundle available (cloud mode network issues or local file missing) |
| BundleExpiredError | Policy bundle has expired and no fresh bundle is available |
| D2PlanLimitError | Account has reached plan limits (tool count, features, or trial expired) |
| TooManyToolsError | Policy exceeds the maximum number of tools allowed for your plan |
| PolicyTooLargeError | Policy bundle exceeds size limits |
| InvalidSignatureError | Policy bundle signature verification failed (cloud mode) |
| ConfigurationError | Invalid configuration or environment setup |
| D2NoContextError | No user context available when calling a protected function |
Production Tip: Catch D2PlanLimitError and BundleExpiredError to implement graceful degradation. For example, log the error and continue with limited functionality rather than crashing.
Always clear context or use context managers. Leaked context can cause security issues where subsequent operations run with wrong user permissions.
# ✅ Recommended: Context manager automatically clears context
with d2.set_user_context("alice", ["admin"]):
result = protected_function()
# Context automatically cleared when exiting block
# ✅ Also safe: Using run_as helper
with d2.run_as("bob", ["viewer"]):
data = get_user_data()✅ Safe: Context is automatically cleared when the block exits.
Recommended: Use with d2.set_user_context() for automatic cleanup.
Alternative: When context managers aren't suitable, you can use d2.set_user() followed by d2.clear_user_context() in a finally block.
⚠️ Never use d2.set_user() without proper cleanup to prevent context leaks.
await async_function()asyncio.create_task(task)anyio.to_thread.run_sync(func)threading.Thread(target=func)multiprocessing.Process()python -m d2 initGenerate policy template with auto-detected functions
--path PATH Custom output path--format FORMAT Output format (yaml, json)--force Overwrite existing policypython -m d2 diagnoseValidate policy structure and limits
python -m d2 inspectPretty-print active bundle and verify signatures
-v, --verbose Show detailed informationpython -m d2 draftUpload policy draft to cloud (requires D2_TOKEN)
python -m d2 publishSign and publish policy for production
python -m d2 pullDownload cloud policy to local file
-o OUTPUT, --output OUTPUT Output file path (default: ./policy.yaml)--format {yaml,json} Force output format when downloading cloud policy (default: yaml)--app-name APP_NAME Specify which app's policy to fetch (overrides local policy file)--stage {published,draft} Which version to fetch: published (stable) or draft (work-in-progress)python -m d2 switch <app>Switch between apps and sync policies
python -m d2 statusShow current app context, bundle info, and cache status
| Variable | Purpose |
|---|---|
| D2_TOKEN | Opaque D2 token (d2_...). Enables cloud mode. |
| D2_POLICY_FILE | Custom path to your policy YAML/JSON file when running in local mode (default: ~/.config/d2/policy.yaml) |
| D2_TELEMETRY | Controls what telemetry data is sent: off (none), metrics (OpenTelemetry only), usage (usage events only), all (everything, default) |
| D2_API_URL | Base URL for the D2 control plane API (default: https://d2.artoo.love) |
| D2_STRICT_SYNC | Set to "1" to raise an error when sync functions are called from async context, instead of automatically running them in a thread pool |
| D2_JWKS_URL | Override JWKS endpoint for signature verification (rare; cloud mode usually auto-discovers) |
| D2_STATE_PATH | Override persisted bundle state path (default: ~/.config/d2/bundles.json); use :memory: to disable persistence |
| D2_SILENT | Set to "1" to suppress startup banner and expiry warnings in local mode |
| OTEL_* | Standard OpenTelemetry configuration variables (e.g., OTEL_EXPORTER_OTLP_ENDPOINT) for metrics export |
D2 provides special threading utilities for cross-thread context management:
from concurrent.futures import ThreadPoolExecutor
import d2
# Use ambient context (snapshot current user)
d2.set_user("alice", ["admin"])
executor = ThreadPoolExecutor()
future = d2.threads.submit_with_context(executor, some_guarded_tool, arg1)
result = future.result() # Runs as alice
# Use explicit actor (preferred for sensitive operations)
actor = d2.UserContext(user_id="bob", roles=frozenset(["viewer"]))
future = d2.threads.submit_with_context(executor, sensitive_tool, actor=actor)
result = future.result() # Runs as bob, regardless of ambient contextCaller controls thread: Submit work to thread pools with proper context isolation.
actor to submit_with_context always overrides any ambient user context. The task runs as the actor, not the submitter.@thread_entrypoint(require_actor=True) enforces that the function can only be called with an explicit actor, rejecting calls that rely on ambient context.Learn about D2's cloud-native authorization platform, policy lifecycle, security model, token management, and dashboard integration.
View System ArchitectureThe D2 Python SDK is open source and available on GitHub:
Enterprise-grade authorization for AI agents