Custom Agent Integration
| Field | Value |
|---|---|
| Document ID | ASCEND-SDK-001 |
| Version | 1.0.0 |
| Last Updated | December 19, 2025 |
| Author | Ascend Engineering Team |
| Publisher | OW-KAI Technologies Inc. |
| Classification | Enterprise Client Documentation |
| Compliance | SOC 2 CC6.1/CC6.2, PCI-DSS 7.1/8.3, HIPAA 164.312, NIST 800-53 AC-2/SI-4 |
Reading Time: 12 minutes | Skill Level: Advanced
Overview
Build AI agents with ASCEND governance from scratch. This guide covers the complete pattern for integrating governance into any custom agent architecture.
Agent Architecture Pattern
┌─────────────────────────────────────────────────────────────────────┐
│ GOVERNED AGENT PATTERN │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ YOUR AGENT │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Input │ │ Reasoning │ │ Action │ │ │
│ │ │ Handler │───▶│ Engine │───▶│ Executor │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ │
│ │ │ │ │
│ └───────────────────────────────────────────────┼──────────────┘ │
│ │ │
│ │ Before │
│ │ Execution │
│ ▼ │
│ ┌─────────────┐ │
│ │ ASCEND │ │
│ │ Governance │ │
│ │ │ │
│ │ ✓ Evaluate │ │
│ │ ✓ Approve │ │
│ │ ✓ Audit │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ External │ │
│ │ Service │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
Base Agent Class
Python
#!/usr/bin/env python3
"""
Base governed agent class for Python.
"""
import os
import logging
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from enum import Enum
from ascend import AscendClient, AgentAction
from ascend.exceptions import (
AuthenticationError,
NetworkError,
TimeoutError
)
logger = logging.getLogger(__name__)
class ActionStatus(Enum):
APPROVED = "approved"
DENIED = "denied"
PENDING = "pending"
ERROR = "error"
@dataclass
class GovernanceResult:
"""Result of governance check."""
status: ActionStatus
action_id: Optional[str] = None
reason: Optional[str] = None
risk_score: Optional[float] = None
risk_level: Optional[str] = None
can_execute: bool = False
class GovernedAgent(ABC):
"""
Base class for AI agents with ASCEND governance.
Subclass this to create governed agents for any use case.
"""
def __init__(
self,
agent_id: str,
agent_name: str,
api_key: Optional[str] = None,
fail_mode: str = "closed"
):
"""
Initialize governed agent.
Args:
agent_id: Unique agent identifier
agent_name: Human-readable name
api_key: ASCEND API key (or env var)
fail_mode: "closed" (deny on error) or "open" (allow on error)
"""
self.agent_id = agent_id
self.agent_name = agent_name
self.fail_mode = fail_mode
self.client = AscendClient(
api_key=api_key or os.environ.get("ASCEND_API_KEY"),
timeout=30
)
# Track current action for audit
self._current_action_id: Optional[str] = None
logger.info(f"Governed agent initialized: {agent_id}")
def check_governance(
self,
action_type: str,
tool_name: str,
description: str,
parameters: Optional[Dict] = None,
context: Optional[Dict] = None,
risk_indicators: Optional[Dict] = None
) -> GovernanceResult:
"""
Check governance before executing an action.
Args:
action_type: Category of action (e.g., "database.query")
tool_name: Tool/service being used
description: Human-readable description
parameters: Action parameters
context: Execution context
risk_indicators: Risk hints
Returns:
GovernanceResult with decision details
"""
action = AgentAction(
agent_id=self.agent_id,
agent_name=self.agent_name,
action_type=action_type,
resource=description,
tool_name=tool_name,
action_details=parameters,
context=context,
risk_indicators=risk_indicators
)
try:
result = self.client.submit_action(action)
self._current_action_id = result.action_id
if result.is_approved():
return GovernanceResult(
status=ActionStatus.APPROVED,
action_id=result.action_id,
risk_score=result.risk_score,
risk_level=result.risk_level,
can_execute=True
)
elif result.is_pending():
return GovernanceResult(
status=ActionStatus.PENDING,
action_id=result.action_id,
reason="Awaiting human approval",
risk_score=result.risk_score,
risk_level=result.risk_level,
can_execute=False
)
else: # denied
return GovernanceResult(
status=ActionStatus.DENIED,
action_id=result.action_id,
reason=result.reason,
risk_score=result.risk_score,
risk_level=result.risk_level,
can_execute=False
)
except (AuthenticationError, NetworkError, TimeoutError) as e:
logger.error(f"Governance check failed: {e}")
if self.fail_mode == "open":
logger.warning("Fail mode OPEN - allowing action")
return GovernanceResult(
status=ActionStatus.APPROVED,
reason="Governance unavailable (fail_mode=open)",
can_execute=True
)
else:
return GovernanceResult(
status=ActionStatus.ERROR,
reason=str(e),
can_execute=False
)
def log_completion(self, result: Optional[Dict] = None):
"""Log successful action completion."""
if self._current_action_id:
try:
self.client.log_action_completed(
self._current_action_id,
result=result
)
except Exception as e:
logger.warning(f"Failed to log completion: {e}")
def log_failure(self, error: str):
"""Log action failure."""
if self._current_action_id:
try:
self.client.log_action_failed(
self._current_action_id,
error={"message": error}
)
except Exception as e:
logger.warning(f"Failed to log failure: {e}")
@abstractmethod
def process(self, input_data: Any) -> Any:
"""
Process input and return output.
Implement this in your agent subclass.
"""
pass
def execute_governed(
self,
action_type: str,
tool_name: str,
description: str,
execute_fn: callable,
parameters: Optional[Dict] = None,
context: Optional[Dict] = None
) -> Any:
"""
Execute a function with governance check.
Args:
action_type: Action category
tool_name: Tool name
description: Action description
execute_fn: Function to execute if approved
parameters: Action parameters
context: Execution context
Returns:
Result of execute_fn if approved
Raises:
PermissionError: If action is denied or pending
"""
# Check governance
gov_result = self.check_governance(
action_type=action_type,
tool_name=tool_name,
description=description,
parameters=parameters,
context=context
)
if not gov_result.can_execute:
raise PermissionError(
f"Action not allowed: {gov_result.reason}"
)
# Execute with audit logging
try:
result = execute_fn()
self.log_completion({"success": True})
return result
except Exception as e:
self.log_failure(str(e))
raise
TypeScript
// governed-agent.ts
import { AscendClient, FailMode } from '@ascend/sdk';
export enum ActionStatus {
APPROVED = 'approved',
DENIED = 'denied',
PENDING = 'pending',
ERROR = 'error'
}
export interface GovernanceResult {
status: ActionStatus;
actionId?: string;
reason?: string;
riskScore?: number;
riskLevel?: string;
canExecute: boolean;
}
export abstract class GovernedAgent {
protected client: AscendClient;
protected agentId: string;
protected agentName: string;
private currentActionId?: string;
constructor(
agentId: string,
agentName: string,
apiKey?: string,
failMode: FailMode = FailMode.CLOSED
) {
this.agentId = agentId;
this.agentName = agentName;
this.client = new AscendClient({
apiKey: apiKey || process.env.ASCEND_API_KEY!,
agentId,
agentName,
failMode
});
}
async checkGovernance(
actionType: string,
toolName: string,
description: string,
parameters?: Record<string, any>,
context?: Record<string, any>
): Promise<GovernanceResult> {
try {
const decision = await this.client.evaluateAction({
actionType,
resource: toolName,
parameters: {
...parameters,
description,
tool_name: toolName
},
context
});
this.currentActionId = decision.actionId;
if (decision.executionAllowed) {
return {
status: ActionStatus.APPROVED,
actionId: decision.actionId,
riskScore: decision.riskScore,
riskLevel: decision.riskLevel,
canExecute: true
};
}
return {
status: decision.decision === 'pending_approval'
? ActionStatus.PENDING
: ActionStatus.DENIED,
actionId: decision.actionId,
reason: decision.reason,
riskScore: decision.riskScore,
riskLevel: decision.riskLevel,
canExecute: false
};
} catch (error) {
console.error('Governance check failed:', error);
return {
status: ActionStatus.ERROR,
reason: error.message,
canExecute: false
};
}
}
async logCompletion(result?: Record<string, any>): Promise<void> {
if (this.currentActionId) {
await this.client.logActionCompleted(this.currentActionId, { result });
}
}
async logFailure(error: string): Promise<void> {
if (this.currentActionId) {
await this.client.logActionFailed(this.currentActionId, {
error: { message: error }
});
}
}
async executeGoverned<T>(
actionType: string,
toolName: string,
description: string,
executeFn: () => Promise<T>,
parameters?: Record<string, any>
): Promise<T> {
const govResult = await this.checkGovernance(
actionType,
toolName,
description,
parameters
);
if (!govResult.canExecute) {
throw new Error(`Action not allowed: ${govResult.reason}`);
}
try {
const result = await executeFn();
await this.logCompletion({ success: true });
return result;
} catch (error) {
await this.logFailure(error.message);
throw error;
}
}
abstract process(input: any): Promise<any>;
}
Example: Data Processing Agent
Python
class DataProcessingAgent(GovernedAgent):
"""Agent that processes customer data with governance."""
def __init__(self):
super().__init__(
agent_id="data-processor-001",
agent_name="Customer Data Processor"
)
self.db = Database()
def process(self, request: Dict) -> Dict:
"""Process a data request."""
operation = request.get("operation")
params = request.get("params", {})
if operation == "query":
return self.query_data(params)
elif operation == "update":
return self.update_data(params)
elif operation == "delete":
return self.delete_data(params)
else:
raise ValueError(f"Unknown operation: {operation}")
def query_data(self, params: Dict) -> Dict:
"""Query customer data with governance."""
return self.execute_governed(
action_type="database.query",
tool_name="postgresql",
description=f"Query {params.get('table')} table",
execute_fn=lambda: self.db.query(params),
parameters=params,
context={"data_classification": "pii"}
)
def update_data(self, params: Dict) -> Dict:
"""Update customer data with governance."""
return self.execute_governed(
action_type="database.update",
tool_name="postgresql",
description=f"Update {params.get('table')} table",
execute_fn=lambda: self.db.update(params),
parameters=params,
context={"data_classification": "pii"}
)
def delete_data(self, params: Dict) -> Dict:
"""Delete customer data with governance (high risk)."""
return self.execute_governed(
action_type="database.delete",
tool_name="postgresql",
description=f"Delete from {params.get('table')} table",
execute_fn=lambda: self.db.delete(params),
parameters={
**params,
"risk_level": "high"
},
context={"data_classification": "pii"}
)
# Usage
agent = DataProcessingAgent()
try:
result = agent.process({
"operation": "query",
"params": {"table": "customers", "limit": 100}
})
print(f"Query result: {result}")
except PermissionError as e:
print(f"Action denied: {e}")
Multi-Action Agent
Handle agents that perform multiple actions in sequence:
class WorkflowAgent(GovernedAgent):
"""Agent that executes multi-step workflows."""
def process(self, workflow: List[Dict]) -> List[Dict]:
"""Execute a workflow with governance for each step."""
results = []
for i, step in enumerate(workflow):
logger.info(f"Executing step {i+1}/{len(workflow)}: {step['name']}")
try:
result = self.execute_governed(
action_type=step["action_type"],
tool_name=step["tool"],
description=step["description"],
execute_fn=lambda s=step: self._execute_step(s),
parameters=step.get("params"),
context={"step": i+1, "total_steps": len(workflow)}
)
results.append({"step": step["name"], "status": "success", "result": result})
except PermissionError as e:
results.append({"step": step["name"], "status": "denied", "reason": str(e)})
if step.get("required", True):
# Stop workflow if required step is denied
break
return results
def _execute_step(self, step: Dict) -> Any:
"""Execute individual workflow step."""
# Implement step execution logic
pass
Registration
Register your agent with ASCEND:
# On agent startup
agent = DataProcessingAgent()
registration = agent.client.register_agent(
agent_type="supervised",
capabilities=["database.query", "database.update", "database.delete"],
allowed_resources=["postgresql"],
metadata={
"version": "1.0.0",
"owner": "data-team@company.com"
}
)
print(f"Registered with trust level: {registration['trust_level']}")
Best Practices
- Fail secure — Default to denying actions on errors
- Log everything — Call
log_completionandlog_failure - Include context — Add user ID, session, environment
- Handle pending — Design for async approval workflows
- Test thoroughly — Test denied and error scenarios
Next Steps
- LangChain Integration — For LangChain agents
- MCP Server — For MCP servers
- Smart Rules — Agent-specific rules
Document Version: 1.0.0 | Last Updated: December 2025