# 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).