- Created examples-javascript.md (20KB, 638 lines) - Authentication, Documents, Governance, Audit examples - Node.js and browser implementations - Complete TractatusClient class - Error handling and retry logic - Rate limiting documentation - Created examples-python.md (30KB, 983 lines) - Authentication, Documents, Governance, Audit examples - Type hints and data classes - Complete TractatusClient class - Error handling decorators and retry logic - Rate limiting documentation - Updated api-reference.html with code examples links - All examples deployed to production Task 12 (API Documentation) - Code examples complete
1152 lines
29 KiB
Markdown
1152 lines
29 KiB
Markdown
# Python API Examples
|
|
|
|
Complete examples for integrating with the Tractatus Framework API using Python with the `requests` library.
|
|
|
|
## Table of Contents
|
|
|
|
- [Installation](#installation)
|
|
- [Authentication](#authentication)
|
|
- [Documents](#documents)
|
|
- [Governance Services](#governance-services)
|
|
- [Audit Logs](#audit-logs)
|
|
- [Error Handling](#error-handling)
|
|
|
|
---
|
|
|
|
## Installation
|
|
|
|
```bash
|
|
pip install requests
|
|
```
|
|
|
|
---
|
|
|
|
## Authentication
|
|
|
|
### Login and Store Token
|
|
|
|
```python
|
|
import requests
|
|
from typing import Dict, Optional
|
|
|
|
API_BASE = "https://agenticgovernance.digital/api"
|
|
# For local development: API_BASE = "http://localhost:9000/api"
|
|
|
|
def login(email: str, password: str) -> Dict:
|
|
"""
|
|
Authenticate and receive JWT token.
|
|
|
|
Args:
|
|
email: User email address
|
|
password: User password
|
|
|
|
Returns:
|
|
dict: Contains 'token' and 'user' keys
|
|
|
|
Raises:
|
|
requests.HTTPError: If authentication fails
|
|
"""
|
|
try:
|
|
response = requests.post(
|
|
f"{API_BASE}/auth/login",
|
|
json={
|
|
"email": email,
|
|
"password": password
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
token = data['token']
|
|
user = data['user']
|
|
|
|
print(f"Login successful: {user['email']}")
|
|
return {'token': token, 'user': user}
|
|
|
|
except requests.HTTPError as e:
|
|
if e.response.status_code == 429:
|
|
print("Too many login attempts. Please wait 15 minutes.")
|
|
elif e.response.status_code == 401:
|
|
print("Invalid credentials")
|
|
else:
|
|
print(f"Login failed: {e}")
|
|
raise
|
|
|
|
|
|
# Usage
|
|
result = login('admin@tractatus.local', 'your_password')
|
|
TOKEN = result['token']
|
|
```
|
|
|
|
### Authenticated Session Class
|
|
|
|
```python
|
|
import requests
|
|
from typing import Dict, Any, Optional
|
|
|
|
class TractatusAPI:
|
|
"""
|
|
Client for interacting with the Tractatus Framework API.
|
|
"""
|
|
|
|
def __init__(self, base_url: str = "https://agenticgovernance.digital/api"):
|
|
self.base_url = base_url
|
|
self.token: Optional[str] = None
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'Content-Type': 'application/json'
|
|
})
|
|
|
|
def login(self, email: str, password: str) -> Dict:
|
|
"""Login and store authentication token."""
|
|
response = self.session.post(
|
|
f"{self.base_url}/auth/login",
|
|
json={"email": email, "password": password}
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.token = data['token']
|
|
|
|
# Update session headers with auth token
|
|
self.session.headers.update({
|
|
'Authorization': f'Bearer {self.token}'
|
|
})
|
|
|
|
return data
|
|
|
|
def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
|
|
"""Make authenticated GET request."""
|
|
if not self.token:
|
|
raise ValueError("Not authenticated. Call login() first.")
|
|
|
|
response = self.session.get(
|
|
f"{self.base_url}{endpoint}",
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def post(self, endpoint: str, data: Dict) -> Dict:
|
|
"""Make authenticated POST request."""
|
|
if not self.token:
|
|
raise ValueError("Not authenticated. Call login() first.")
|
|
|
|
response = self.session.post(
|
|
f"{self.base_url}{endpoint}",
|
|
json=data
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'your_password')
|
|
|
|
# Now make authenticated requests
|
|
status = client.get('/governance/status')
|
|
print(status)
|
|
```
|
|
|
|
---
|
|
|
|
## Documents
|
|
|
|
### List All Documents
|
|
|
|
```python
|
|
def list_documents(
|
|
page: int = 1,
|
|
limit: int = 50,
|
|
quadrant: Optional[str] = None
|
|
) -> Dict:
|
|
"""
|
|
Retrieve list of documents with optional filtering.
|
|
|
|
Args:
|
|
page: Page number (default: 1)
|
|
limit: Results per page (default: 50)
|
|
quadrant: Filter by quadrant (STRATEGIC, OPERATIONAL, etc.)
|
|
|
|
Returns:
|
|
dict: Contains 'documents' array and 'pagination' info
|
|
"""
|
|
params = {
|
|
'page': page,
|
|
'limit': limit
|
|
}
|
|
|
|
if quadrant:
|
|
params['quadrant'] = quadrant
|
|
|
|
response = requests.get(
|
|
f"{API_BASE}/documents",
|
|
params=params
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return data
|
|
|
|
|
|
# Usage
|
|
result = list_documents(page=1, limit=10, quadrant='STRATEGIC')
|
|
print(f"Found {result['pagination']['total']} documents")
|
|
|
|
for doc in result['documents']:
|
|
print(f"- {doc['title']} ({doc['quadrant']})")
|
|
```
|
|
|
|
### Get Single Document
|
|
|
|
```python
|
|
def get_document(identifier: str) -> Dict:
|
|
"""
|
|
Retrieve a single document by ID or slug.
|
|
|
|
Args:
|
|
identifier: Document MongoDB ObjectId or URL slug
|
|
|
|
Returns:
|
|
dict: Document data
|
|
|
|
Raises:
|
|
requests.HTTPError: If document not found (404)
|
|
"""
|
|
response = requests.get(f"{API_BASE}/documents/{identifier}")
|
|
|
|
if response.status_code == 404:
|
|
raise ValueError(f"Document not found: {identifier}")
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data['document']
|
|
|
|
|
|
# Usage (by slug)
|
|
doc = get_document('introduction-to-tractatus')
|
|
print(f"Title: {doc['title']}")
|
|
print(f"Quadrant: {doc['quadrant']}")
|
|
|
|
# Usage (by ID)
|
|
doc = get_document('672f821b6e820c0c7a0e0d55')
|
|
print(doc)
|
|
```
|
|
|
|
### Search Documents
|
|
|
|
```python
|
|
def search_documents(query: str) -> Dict:
|
|
"""
|
|
Full-text search across all documents.
|
|
|
|
Args:
|
|
query: Search query string
|
|
|
|
Returns:
|
|
dict: Contains 'results' array and 'count'
|
|
"""
|
|
response = requests.get(
|
|
f"{API_BASE}/documents/search",
|
|
params={'q': query}
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return data
|
|
|
|
|
|
# Usage
|
|
results = search_documents('boundary enforcement')
|
|
print(f"Found {results['count']} results")
|
|
|
|
for result in results['results']:
|
|
print(f"- {result['title']} (score: {result['score']:.2f})")
|
|
if 'excerpt' in result:
|
|
print(f" Excerpt: {result['excerpt'][:100]}...")
|
|
```
|
|
|
|
### Create Document (Admin Only)
|
|
|
|
```python
|
|
def create_document(
|
|
client: TractatusAPI,
|
|
title: str,
|
|
slug: str,
|
|
quadrant: str,
|
|
content: str,
|
|
status: str = 'published'
|
|
) -> Dict:
|
|
"""
|
|
Create a new framework document (requires admin authentication).
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client
|
|
title: Document title
|
|
slug: URL slug (lowercase, hyphens only)
|
|
quadrant: One of: STRATEGIC, OPERATIONAL, TACTICAL, SYSTEM, STOCHASTIC
|
|
content: Document content in Markdown format
|
|
status: One of: draft, published, archived (default: published)
|
|
|
|
Returns:
|
|
dict: Created document
|
|
|
|
Raises:
|
|
requests.HTTPError: If creation fails (403 = forbidden, 409 = slug exists)
|
|
"""
|
|
document_data = {
|
|
'title': title,
|
|
'slug': slug,
|
|
'quadrant': quadrant,
|
|
'content_markdown': content,
|
|
'status': status
|
|
}
|
|
|
|
try:
|
|
response = client.post('/documents', document_data)
|
|
doc = response['document']
|
|
print(f"Document created: {doc['_id']}")
|
|
return doc
|
|
|
|
except requests.HTTPError as e:
|
|
if e.response.status_code == 403:
|
|
print("Error: Admin role required")
|
|
elif e.response.status_code == 409:
|
|
print("Error: Slug already exists")
|
|
raise
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
new_doc = create_document(
|
|
client=client,
|
|
title='Advanced Boundary Enforcement Patterns',
|
|
slug='advanced-boundary-enforcement',
|
|
quadrant='OPERATIONAL',
|
|
content='# Advanced Patterns\n\nThis document explores...',
|
|
status='published'
|
|
)
|
|
```
|
|
|
|
---
|
|
|
|
## Governance Services
|
|
|
|
### InstructionPersistenceClassifier
|
|
|
|
```python
|
|
def classify_instruction(
|
|
client: TractatusAPI,
|
|
text: str,
|
|
context: Optional[Dict] = None
|
|
) -> Dict:
|
|
"""
|
|
Classify an instruction by quadrant and persistence level.
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
text: Instruction text to classify
|
|
context: Optional context (source, session_id, etc.)
|
|
|
|
Returns:
|
|
dict: Classification with quadrant, persistence, temporal_scope,
|
|
verification_required, reasoning, and confidence
|
|
"""
|
|
if context is None:
|
|
context = {}
|
|
|
|
context.setdefault('source', 'user')
|
|
context.setdefault('session_id', 'default')
|
|
|
|
response = client.post('/governance/classify', {
|
|
'text': text,
|
|
'context': context
|
|
})
|
|
|
|
return response['classification']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
classification = classify_instruction(
|
|
client,
|
|
'Always use MongoDB on port 27027',
|
|
{'source': 'user', 'session_id': 'sess_123'}
|
|
)
|
|
|
|
print(f"Quadrant: {classification['quadrant']}")
|
|
print(f"Persistence: {classification['persistence']}")
|
|
print(f"Temporal Scope: {classification['temporal_scope']}")
|
|
print(f"Confidence: {classification['confidence']:.2%}")
|
|
print(f"Reasoning: {classification['reasoning']}")
|
|
```
|
|
|
|
### CrossReferenceValidator
|
|
|
|
```python
|
|
def validate_action(
|
|
client: TractatusAPI,
|
|
action: Dict,
|
|
context: Optional[Dict] = None
|
|
) -> Dict:
|
|
"""
|
|
Validate a proposed action against instruction history.
|
|
|
|
Detects conflicts and training pattern overrides (27027 failure mode).
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
action: Action to validate (type, target, parameters, etc.)
|
|
context: Optional context (messages, session_id, etc.)
|
|
|
|
Returns:
|
|
dict: Validation result with status, conflicts, and recommendation
|
|
"""
|
|
if context is None:
|
|
context = {}
|
|
|
|
context.setdefault('messages', [])
|
|
context.setdefault('session_id', 'default')
|
|
|
|
response = client.post('/governance/validate', {
|
|
'action': action,
|
|
'context': context
|
|
})
|
|
|
|
return response['validation']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
action = {
|
|
'type': 'database_config',
|
|
'target': 'MongoDB',
|
|
'parameters': {'port': 27017}
|
|
}
|
|
|
|
validation = validate_action(client, action)
|
|
|
|
if validation['status'] == 'REJECTED':
|
|
print("❌ Action rejected")
|
|
print(f"Reason: {validation['reason']}")
|
|
|
|
for conflict in validation.get('conflicts', []):
|
|
print(f" Conflicts with: {conflict['text']} ({conflict['instruction_id']})")
|
|
|
|
print(f"Recommendation: {validation['recommendation']}")
|
|
|
|
elif validation['status'] == 'APPROVED':
|
|
print("✅ Action approved")
|
|
|
|
elif validation['status'] == 'WARNING':
|
|
print("⚠️ Action has warnings")
|
|
```
|
|
|
|
### BoundaryEnforcer
|
|
|
|
```python
|
|
def enforce_boundary(
|
|
client: TractatusAPI,
|
|
action: Dict,
|
|
context: Optional[Dict] = None
|
|
) -> Dict:
|
|
"""
|
|
Check if an action crosses into values territory requiring human approval.
|
|
|
|
Boundaries: privacy, ethics, sovereignty, strategic
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
action: Action to check (type, description, impact, etc.)
|
|
context: Optional context
|
|
|
|
Returns:
|
|
dict: Enforcement with decision (ALLOW/BLOCK/ESCALATE), boundary,
|
|
reasoning, alternatives, and requiresHuman flag
|
|
"""
|
|
if context is None:
|
|
context = {}
|
|
|
|
response = client.post('/governance/enforce', {
|
|
'action': action,
|
|
'context': context
|
|
})
|
|
|
|
return response['enforcement']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
action = {
|
|
'type': 'policy_change',
|
|
'description': 'Update privacy policy to enable more tracking',
|
|
'impact': 'user_privacy'
|
|
}
|
|
|
|
enforcement = enforce_boundary(client, action)
|
|
|
|
if enforcement['decision'] == 'BLOCK':
|
|
print("🚫 Action blocked - crosses values boundary")
|
|
print(f"Boundary: {enforcement['boundary_crossed']}")
|
|
print(f"Reason: {enforcement['reason']}")
|
|
|
|
print("\nAlternatives:")
|
|
for i, alt in enumerate(enforcement['alternatives'], 1):
|
|
print(f"{i}. {alt}")
|
|
|
|
elif enforcement['decision'] == 'ALLOW':
|
|
print("✅ Action allowed")
|
|
|
|
elif enforcement['decision'] == 'ESCALATE':
|
|
print("⚠️ Action requires escalation")
|
|
```
|
|
|
|
### ContextPressureMonitor
|
|
|
|
```python
|
|
def analyze_pressure(
|
|
client: TractatusAPI,
|
|
context: Dict
|
|
) -> Dict:
|
|
"""
|
|
Analyze session context pressure across multiple factors.
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
context: Session context with tokenUsage, messageCount, errorCount, etc.
|
|
|
|
Returns:
|
|
dict: Pressure analysis with level (NORMAL/ELEVATED/HIGH/CRITICAL/DANGEROUS),
|
|
score, factors, recommendation, and triggerHandoff flag
|
|
"""
|
|
response = client.post('/governance/pressure', {
|
|
'context': context
|
|
})
|
|
|
|
return response['pressure']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
context = {
|
|
'tokenUsage': 120000,
|
|
'tokenBudget': 200000,
|
|
'messageCount': 45,
|
|
'errorCount': 3,
|
|
'complexOperations': 8,
|
|
'sessionDuration': 3600
|
|
}
|
|
|
|
pressure = analyze_pressure(client, context)
|
|
|
|
print(f"Pressure Level: {pressure['level']}")
|
|
print(f"Score: {pressure['score']}%")
|
|
|
|
print("\nFactors:")
|
|
for factor, data in pressure['factors'].items():
|
|
print(f" {factor}: {data['value']} ({data['status']})")
|
|
|
|
print(f"\nRecommendation: {pressure['recommendation']}")
|
|
|
|
if pressure.get('triggerHandoff'):
|
|
print("⚠️ Session handoff recommended")
|
|
|
|
if pressure.get('next_checkpoint'):
|
|
print(f"Next checkpoint at: {pressure['next_checkpoint']} tokens")
|
|
```
|
|
|
|
### MetacognitiveVerifier
|
|
|
|
```python
|
|
def verify_action(
|
|
client: TractatusAPI,
|
|
action: Dict,
|
|
reasoning: Dict,
|
|
context: Optional[Dict] = None
|
|
) -> Dict:
|
|
"""
|
|
Perform metacognitive verification on proposed action.
|
|
|
|
Detects scope creep, misalignment, and provides confidence scoring.
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
action: Action to verify (type, scope, complexity, etc.)
|
|
reasoning: Reasoning for the action (intent, approach, risks, etc.)
|
|
context: Optional context (requested, original_scope, etc.)
|
|
|
|
Returns:
|
|
dict: Verification with decision (APPROVED/REQUIRE_REVIEW/REJECTED),
|
|
confidence, concerns, criteria scores, alternatives, and scopeCreep flag
|
|
"""
|
|
if context is None:
|
|
context = {}
|
|
|
|
response = client.post('/governance/verify', {
|
|
'action': action,
|
|
'reasoning': reasoning,
|
|
'context': context
|
|
})
|
|
|
|
return response['verification']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
action = {
|
|
'type': 'refactor',
|
|
'scope': 'Refactor 47 files across 5 system areas',
|
|
'complexity': 'high'
|
|
}
|
|
|
|
reasoning = {
|
|
'intent': 'Improve code organization',
|
|
'approach': 'Extract shared utilities, consolidate duplicates',
|
|
'risks': 'Potential breaking changes'
|
|
}
|
|
|
|
context = {
|
|
'requested': 'Refactor authentication module',
|
|
'original_scope': 'single module'
|
|
}
|
|
|
|
verification = verify_action(client, action, reasoning, context)
|
|
|
|
print(f"Decision: {verification['decision']}")
|
|
print(f"Confidence: {verification['confidence']:.2%}")
|
|
|
|
if verification['concerns']:
|
|
print("\n⚠️ Concerns:")
|
|
for concern in verification['concerns']:
|
|
print(f" [{concern['severity']}] {concern['type']}: {concern['detail']}")
|
|
|
|
if verification.get('scopeCreep'):
|
|
print("\n🔴 Scope creep detected")
|
|
|
|
print("\nCriteria Scores:")
|
|
for criterion, score in verification['criteria'].items():
|
|
print(f" {criterion}: {score * 100:.0f}%")
|
|
|
|
if verification.get('alternatives'):
|
|
print("\nAlternatives:")
|
|
for i, alt in enumerate(verification['alternatives'], 1):
|
|
print(f"{i}. {alt}")
|
|
```
|
|
|
|
---
|
|
|
|
## Audit Logs
|
|
|
|
### Get Audit Logs with Filtering
|
|
|
|
```python
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Optional
|
|
|
|
def get_audit_logs(
|
|
client: TractatusAPI,
|
|
page: int = 1,
|
|
limit: int = 50,
|
|
action: Optional[str] = None,
|
|
user_id: Optional[str] = None,
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None
|
|
) -> Dict:
|
|
"""
|
|
Retrieve audit logs with filtering and pagination.
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
page: Page number (default: 1)
|
|
limit: Results per page (default: 50, max: 100)
|
|
action: Filter by action type
|
|
user_id: Filter by user ID
|
|
start_date: Filter by start date
|
|
end_date: Filter by end date
|
|
|
|
Returns:
|
|
dict: Contains 'logs' array, 'total', and pagination info
|
|
"""
|
|
params = {
|
|
'page': page,
|
|
'limit': limit
|
|
}
|
|
|
|
if action:
|
|
params['action'] = action
|
|
if user_id:
|
|
params['userId'] = user_id
|
|
if start_date:
|
|
params['startDate'] = start_date.isoformat()
|
|
if end_date:
|
|
params['endDate'] = end_date.isoformat()
|
|
|
|
response = client.get('/audit/audit-logs', params=params)
|
|
return response
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
# Get logs from the last 7 days
|
|
start_date = datetime.now() - timedelta(days=7)
|
|
logs_data = get_audit_logs(
|
|
client,
|
|
page=1,
|
|
limit=20,
|
|
action='validate_action',
|
|
start_date=start_date
|
|
)
|
|
|
|
print(f"Total logs: {logs_data['total']}")
|
|
|
|
for log in logs_data['logs']:
|
|
timestamp = log['timestamp']
|
|
service = log['service']
|
|
action = log['action']
|
|
status = log['status']
|
|
|
|
print(f"[{timestamp}] {service}: {action} - {status}")
|
|
|
|
if log.get('details'):
|
|
import json
|
|
print(f" Details: {json.dumps(log['details'], indent=2)}")
|
|
```
|
|
|
|
### Get Audit Analytics
|
|
|
|
```python
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
def get_audit_analytics(
|
|
client: TractatusAPI,
|
|
start_date: Optional[datetime] = None,
|
|
end_date: Optional[datetime] = None
|
|
) -> Dict:
|
|
"""
|
|
Get aggregated analytics on audit activity.
|
|
|
|
Args:
|
|
client: Authenticated TractatusAPI client (admin)
|
|
start_date: Start date for analytics period
|
|
end_date: End date for analytics period
|
|
|
|
Returns:
|
|
dict: Analytics with total_events, by_service, by_status,
|
|
rejection_rate, and period information
|
|
"""
|
|
params = {}
|
|
|
|
if start_date:
|
|
params['startDate'] = start_date.isoformat()
|
|
if end_date:
|
|
params['endDate'] = end_date.isoformat()
|
|
|
|
response = client.get('/audit/audit-analytics', params=params)
|
|
return response['analytics']
|
|
|
|
|
|
# Usage
|
|
client = TractatusAPI()
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
# Get analytics for October 2025
|
|
analytics = get_audit_analytics(
|
|
client,
|
|
start_date=datetime(2025, 10, 1),
|
|
end_date=datetime(2025, 10, 31)
|
|
)
|
|
|
|
print(f"Total Events: {analytics['total_events']}")
|
|
|
|
print("\nBreakdown by Service:")
|
|
for service, count in analytics['by_service'].items():
|
|
print(f" {service}: {count}")
|
|
|
|
print("\nBreakdown by Status:")
|
|
for status, count in analytics['by_status'].items():
|
|
print(f" {status}: {count}")
|
|
|
|
print(f"\nRejection Rate: {analytics['rejection_rate']}%")
|
|
|
|
period = analytics['period']
|
|
print(f"\nPeriod: {period['start']} to {period['end']} ({period['days']} days)")
|
|
```
|
|
|
|
---
|
|
|
|
## Error Handling
|
|
|
|
### Comprehensive Error Handler
|
|
|
|
```python
|
|
import requests
|
|
from typing import Callable, Any
|
|
|
|
def handle_api_errors(func: Callable) -> Callable:
|
|
"""
|
|
Decorator for handling API errors consistently.
|
|
"""
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
|
|
except requests.HTTPError as e:
|
|
status = e.response.status_code
|
|
data = e.response.json() if e.response.text else {}
|
|
|
|
error_handlers = {
|
|
400: lambda: print(f"Bad Request: {data.get('message', 'Invalid input')}"),
|
|
401: lambda: print("Unauthorized: Please login"),
|
|
403: lambda: print(f"Forbidden: {data.get('message', 'Insufficient permissions')}"),
|
|
404: lambda: print(f"Not Found: {data.get('message', 'Resource not found')}"),
|
|
409: lambda: print(f"Conflict: {data.get('message', 'Resource already exists')}"),
|
|
429: lambda: print(f"Rate Limit Exceeded: {data.get('message')}"),
|
|
500: lambda: print(f"Internal Server Error: {data.get('errorId', 'Unknown')}")
|
|
}
|
|
|
|
handler = error_handlers.get(status, lambda: print(f"API Error {status}: {data.get('message')}"))
|
|
handler()
|
|
|
|
raise
|
|
|
|
except requests.ConnectionError:
|
|
print("Network Error: Unable to connect to API")
|
|
print("Check your internet connection and API base URL")
|
|
raise
|
|
|
|
except requests.Timeout:
|
|
print("Request Timeout: API did not respond in time")
|
|
raise
|
|
|
|
except Exception as e:
|
|
print(f"Unexpected Error: {type(e).__name__}: {e}")
|
|
raise
|
|
|
|
return wrapper
|
|
|
|
|
|
# Usage
|
|
@handle_api_errors
|
|
def get_document_safe(identifier: str) -> Dict:
|
|
return get_document(identifier)
|
|
|
|
|
|
doc = get_document_safe('some-slug')
|
|
```
|
|
|
|
### Retry Logic with Exponential Backoff
|
|
|
|
```python
|
|
import time
|
|
import requests
|
|
from typing import Callable, Any
|
|
|
|
def retry_with_backoff(
|
|
func: Callable,
|
|
max_retries: int = 3,
|
|
base_delay: float = 1.0
|
|
) -> Any:
|
|
"""
|
|
Retry function with exponential backoff.
|
|
|
|
Args:
|
|
func: Function to retry
|
|
max_retries: Maximum number of retry attempts
|
|
base_delay: Base delay in seconds (doubles each retry)
|
|
|
|
Returns:
|
|
Result of successful function call
|
|
|
|
Raises:
|
|
Exception: If all retries fail
|
|
"""
|
|
for attempt in range(1, max_retries + 1):
|
|
try:
|
|
return func()
|
|
|
|
except requests.HTTPError as e:
|
|
# Don't retry on client errors (4xx except 429)
|
|
if 400 <= e.response.status_code < 500 and e.response.status_code != 429:
|
|
raise
|
|
|
|
if attempt == max_retries:
|
|
raise
|
|
|
|
delay = base_delay * (2 ** (attempt - 1))
|
|
print(f"Attempt {attempt} failed. Retrying in {delay}s...")
|
|
time.sleep(delay)
|
|
|
|
except (requests.ConnectionError, requests.Timeout) as e:
|
|
if attempt == max_retries:
|
|
raise
|
|
|
|
delay = base_delay * (2 ** (attempt - 1))
|
|
print(f"Network error. Retrying in {delay}s...")
|
|
time.sleep(delay)
|
|
|
|
|
|
# Usage
|
|
def fetch_document():
|
|
return get_document('some-slug')
|
|
|
|
doc = retry_with_backoff(fetch_document, max_retries=3, base_delay=1.0)
|
|
```
|
|
|
|
---
|
|
|
|
## Complete Example: Full Integration
|
|
|
|
```python
|
|
import requests
|
|
from typing import Dict, Optional, Any
|
|
from datetime import datetime
|
|
|
|
class TractatusClient:
|
|
"""
|
|
Complete client for Tractatus Framework API.
|
|
"""
|
|
|
|
def __init__(self, base_url: str = "https://agenticgovernance.digital/api"):
|
|
self.base_url = base_url
|
|
self.token: Optional[str] = None
|
|
self.session = requests.Session()
|
|
self.session.headers.update({'Content-Type': 'application/json'})
|
|
|
|
def login(self, email: str, password: str) -> Dict:
|
|
"""Authenticate and store token."""
|
|
response = self.session.post(
|
|
f"{self.base_url}/auth/login",
|
|
json={"email": email, "password": password}
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.token = data['token']
|
|
self.session.headers.update({'Authorization': f'Bearer {self.token}'})
|
|
|
|
print(f"✅ Logged in as: {data['user']['email']}")
|
|
return data
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> Dict:
|
|
"""Make authenticated request."""
|
|
if not self.token:
|
|
raise ValueError("Not authenticated. Call login() first.")
|
|
|
|
response = self.session.request(
|
|
method,
|
|
f"{self.base_url}{endpoint}",
|
|
**kwargs
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_documents(self, **params) -> Dict:
|
|
"""List documents."""
|
|
return self._request('GET', '/documents', params=params)
|
|
|
|
def get_document(self, identifier: str) -> Dict:
|
|
"""Get single document."""
|
|
return self._request('GET', f'/documents/{identifier}')
|
|
|
|
def classify_instruction(self, text: str, context: Optional[Dict] = None) -> Dict:
|
|
"""Classify instruction."""
|
|
return self._request('POST', '/governance/classify', json={
|
|
'text': text,
|
|
'context': context or {}
|
|
})
|
|
|
|
def validate_action(self, action: Dict, context: Optional[Dict] = None) -> Dict:
|
|
"""Validate action."""
|
|
return self._request('POST', '/governance/validate', json={
|
|
'action': action,
|
|
'context': context or {}
|
|
})
|
|
|
|
def enforce_boundary(self, action: Dict, context: Optional[Dict] = None) -> Dict:
|
|
"""Check boundary enforcement."""
|
|
return self._request('POST', '/governance/enforce', json={
|
|
'action': action,
|
|
'context': context or {}
|
|
})
|
|
|
|
def analyze_pressure(self, context: Dict) -> Dict:
|
|
"""Analyze context pressure."""
|
|
return self._request('POST', '/governance/pressure', json={'context': context})
|
|
|
|
def verify_action(self, action: Dict, reasoning: Dict, context: Optional[Dict] = None) -> Dict:
|
|
"""Metacognitive verification."""
|
|
return self._request('POST', '/governance/verify', json={
|
|
'action': action,
|
|
'reasoning': reasoning,
|
|
'context': context or {}
|
|
})
|
|
|
|
def get_audit_logs(self, **params) -> Dict:
|
|
"""Get audit logs."""
|
|
return self._request('GET', '/audit/audit-logs', params=params)
|
|
|
|
def get_audit_analytics(self, **params) -> Dict:
|
|
"""Get audit analytics."""
|
|
return self._request('GET', '/audit/audit-analytics', params=params)
|
|
|
|
|
|
# Usage Example
|
|
def main():
|
|
# Initialize client
|
|
client = TractatusClient()
|
|
|
|
# Login
|
|
client.login('admin@tractatus.local', 'password')
|
|
|
|
# Classify an instruction
|
|
print("\n📋 Classifying instruction...")
|
|
classification = client.classify_instruction(
|
|
'Always use MongoDB on port 27027'
|
|
)
|
|
print(f"Quadrant: {classification['classification']['quadrant']}")
|
|
print(f"Persistence: {classification['classification']['persistence']}")
|
|
|
|
# Validate an action
|
|
print("\n✅ Validating action...")
|
|
validation = client.validate_action({
|
|
'type': 'database_config',
|
|
'target': 'MongoDB',
|
|
'parameters': {'port': 27017}
|
|
})
|
|
print(f"Status: {validation['validation']['status']}")
|
|
|
|
# Check boundary enforcement
|
|
print("\n🚧 Checking boundary...")
|
|
enforcement = client.enforce_boundary({
|
|
'type': 'policy_change',
|
|
'description': 'Update privacy policy',
|
|
'impact': 'user_privacy'
|
|
})
|
|
print(f"Decision: {enforcement['enforcement']['decision']}")
|
|
|
|
# Analyze pressure
|
|
print("\n📊 Analyzing pressure...")
|
|
pressure = client.analyze_pressure({
|
|
'tokenUsage': 50000,
|
|
'tokenBudget': 200000,
|
|
'messageCount': 20
|
|
})
|
|
print(f"Level: {pressure['pressure']['level']}")
|
|
|
|
# Get recent documents
|
|
print("\n📚 Fetching documents...")
|
|
docs = client.get_documents(limit=5)
|
|
print(f"Found {docs['pagination']['total']} total documents")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
```
|
|
|
|
---
|
|
|
|
## Rate Limiting
|
|
|
|
The Tractatus API implements rate limiting:
|
|
|
|
- **Login endpoint**: 5 attempts per 15 minutes per IP
|
|
- **General API**: 100 requests per 15 minutes per IP
|
|
|
|
Handle rate limiting:
|
|
|
|
```python
|
|
import time
|
|
import requests
|
|
|
|
def api_call_with_rate_limit(func):
|
|
"""Handle rate limiting with automatic retry."""
|
|
try:
|
|
return func()
|
|
except requests.HTTPError as e:
|
|
if e.response.status_code == 429:
|
|
retry_after = int(e.response.headers.get('Retry-After', 60))
|
|
print(f"⚠️ Rate limited. Waiting {retry_after} seconds...")
|
|
time.sleep(retry_after)
|
|
return func()
|
|
raise
|
|
|
|
|
|
# Usage
|
|
result = api_call_with_rate_limit(lambda: get_document('some-slug'))
|
|
```
|
|
|
|
---
|
|
|
|
## Type Hints and Data Classes
|
|
|
|
For better type safety, use Python data classes:
|
|
|
|
```python
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
from enum import Enum
|
|
|
|
class Quadrant(Enum):
|
|
STRATEGIC = "STRATEGIC"
|
|
OPERATIONAL = "OPERATIONAL"
|
|
TACTICAL = "TACTICAL"
|
|
SYSTEM = "SYSTEM"
|
|
STOCHASTIC = "STOCHASTIC"
|
|
|
|
class Persistence(Enum):
|
|
HIGH = "HIGH"
|
|
MEDIUM = "MEDIUM"
|
|
LOW = "LOW"
|
|
|
|
class PressureLevel(Enum):
|
|
NORMAL = "NORMAL"
|
|
ELEVATED = "ELEVATED"
|
|
HIGH = "HIGH"
|
|
CRITICAL = "CRITICAL"
|
|
DANGEROUS = "DANGEROUS"
|
|
|
|
@dataclass
|
|
class Classification:
|
|
quadrant: Quadrant
|
|
persistence: Persistence
|
|
temporal_scope: str
|
|
verification_required: str
|
|
reasoning: str
|
|
confidence: float
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
status: str
|
|
reason: Optional[str] = None
|
|
conflicts: List[Dict] = None
|
|
recommendation: Optional[str] = None
|
|
|
|
@dataclass
|
|
class PressureAnalysis:
|
|
level: PressureLevel
|
|
score: float
|
|
factors: Dict
|
|
recommendation: str
|
|
triggerHandoff: bool
|
|
next_checkpoint: Optional[int] = None
|
|
```
|
|
|
|
---
|
|
|
|
For more information, see the [API Reference](https://agenticgovernance.digital/api-reference.html) and [OpenAPI Specification](https://agenticgovernance.digital/docs/api/openapi.yaml).
|