Enterprise AI Agent Three-Step Practical Implementation Framework

# Template: <system> <action> --param1=value --param2=value
# Practical Examples
bpm approve --instanceId=xxx --user=zhangsan --comment="Compliant approval passed"
mail scan --today --tag=contract_approval
feishu notify --chatId=group_001 --msg="Approval process completed"
crm update --entity=lead --id=12345 --fields='{"status":"approved"}'

from abc import ABC
from typing import Dict, Any
import logging

class CLIGateway:
    def __init__(self):
        self.handlers: Dict[str, Any] = {}
        self.auth_proxy = IdentityAuthProxy()
        self.permission_control = PermissionControl()
        self.rate_limiter = RateLimiter()
        self.audit_logger = AuditLogger()

    def register(self, system: str, handler: Any):
        """Register system-specific command handlers"""
        self.handlers[system] = handler

    def execute(self, command: str, user_id: str, **kwargs) -> Dict:
        # 1. Verify user identity
        identity = self.auth_proxy.verify(user_id)
        # 2. Validate operation permissions
        if not self.permission_control.check(command, identity):
            return {"error": "permission_denied"}
        # 3. Enforce rate limits
        if not self.rate_limiter.allow(command):
            return {"error": "rate_limit_exceeded"}
        # 4. Dispatch command to target system
        result = self._dispatch(command, kwargs)
        # 5. Log operation for audit
        self.audit_logger.log(command, user_id, result)
        return result

    def _dispatch(self, command: str, params: Dict) -> Dict:
        system, action = command.split()[:2]
        return self.handlers[system].execute(action, params) if system in self.handlers else {"error": "system_not_found"}

from openai import OpenAI

class BPMCLI:
    def __init__(self, api_base: str, api_key: str):
        self.client = OpenAI(api_key=api_key, base_url=api_base)

    def approve(self, instance_id: str, user: str, comment: str) -> Dict:
        """Process BPM approval requests"""
        return self._call("bpm.approve", {"instance_id": instance_id, "user": user, "comment": comment})

    def query_status(self, instance_id: str) -> Dict:
        """Check BPM approval status"""
        return self._call("bpm.status", {"instance_id": instance_id})

    def _call(self, action: str, params: Dict) -> Dict:
        # Integrate with actual BPM system API
        pass

from typing import Optional, List

class FeishuCLI:
    def notify(self, chat_id: str, msg: str, at: Optional[List[str]] = None) -> Dict:
        """Send group notifications via Feishu"""
        return self._call("feishu.notify", {"chat_id": chat_id, "msg": msg, "at": at or []})

    def _call(self, action: str, params: Dict) -> Dict:
        # Integrate with Feishu API
        pass

Skill = Trigger Conditions + Rule Engine + CLI Orchestration + Exception Handling + Feedback Mechanism

from abc import ABC, abstractmethod
from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class SkillConfig:
    name: str
    description: str
    trigger_conditions: List[Dict]
    rules: Dict[str, Any]
    required_cli: List[str]

class BaseSkill(ABC):
    def __init__(self, config: SkillConfig, cli_gateway: CLIGateway):
        self.config = config
        self.cli = cli_gateway

    @abstractmethod
    def should_trigger(self, context: Dict) -> bool:
        """Determine if the skill should activate based on context"""
        pass

    @abstractmethod
    def execute(self, context: Dict) -> Dict:
        """Execute the skill workflow"""
        pass

    def handle_exception(self, error: Exception, context: Dict) -> Dict:
        """Handle workflow errors gracefully"""
        return {"status": "error", "message": str(error)}

from skills.base_skill import BaseSkill, SkillConfig

class ContractApprovalSkill(BaseSkill):
    def __init__(self, cli_gateway: CLIGateway):
        config = SkillConfig(
            name="contract_approval",
            description="Automated contract review and approval workflow",
            trigger_conditions=[
                {"type": "email", "keywords": ["contract", "agreement"]},
                {"type": "attachment", "extensions": ["pdf", "docx"]}
            ],
            rules={
                "amount_threshold": 100000,
                "high_risk_clauses": ["guarantee", "indemnity", "joint_liability"],
                "extra_sign_threshold": 500000
            },
            required_cli=["mail", "doc", "bpm", "feishu"]
        )
        super().__init__(config, cli_gateway)

    def should_trigger(self, context: Dict) -> bool:
        """Check if context meets trigger conditions"""
        content = context.get("content", "")
        return any(kw in content for cond in self.config.trigger_conditions for kw in cond.get("keywords", []))

    def execute(self, context: Dict) -> Dict:
        try:
            # 1. Validate user permissions
            if not self._check_permission(context):
                return {"status": "denied", "reason": "insufficient_permissions"}

            # 2. Analyze contract risk via LLM
            risk_result = self._analyze_contract_risk(context)
            if risk_result["level"] == "high":
                self._route_to_human_review(context)
                return {"status": "pending_review", "reason": "high_risk_contract"}

            # 3. Execute approval workflow via CLI
            self.cli.execute("bpm approve", **risk_result["approval_params"])
            self.cli.execute("feishu notify", **risk_result["notification_params"])
            return {"status": "completed", "decision": risk_result}

        except Exception as e:
            return self.handle_exception(e, context)

    def _analyze_contract_risk(self, context: Dict) -> Dict:
        """Call LLM for contract risk analysis via unified gateway"""
        prompt = f"Analyze risk for contract content: {context.get('content', '')}"
        response = self.llm_call(prompt, model="claude-4.6")
        return self._parse_risk_response(response)

    def llm_call(self, prompt: str, model: str) -> str:
        """LLM integration via 4sapi gateway"""
        pass

from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class EmployeeContext:
    employee_id: str
    department: str
    role: str
    permissions: List[str]
    bound_skills: List[str]

class EmployeeAgent:
    def __init__(self, context: EmployeeContext, skill_library, cli_gateway, llm_client):
        self.context = context
        self.skills = skill_library
        self.cli = cli_gateway
        self.llm = llm_client

    def process(self, event: Dict) -> Dict:
        """Process business events via assigned skills"""
        # 1. Detect event type
        event_type = self._detect_event(event)
        # 2. Check if Agent should handle the event
        if not self._is_eligible(event_type):
            return {"status": "ignored"}
        # 3. Match relevant skill
        skill = self.skills.find(event_type, self.context.bound_skills)
        if not skill:
            return {"status": "no_matching_skill"}
        # 4. Execute skill
        return skill.execute(event)

employee_agents:
  legal_agent:
    employee_id: L001
    department: Legal
    role: Legal Specialist
    bound_skills: [contract_approval, contract_archival]
    permissions: [mail.read, bpm.approve, feishu.notify]
  finance_agent:
    employee_id: F001
    department: Finance
    role: Finance Specialist
    bound_skills: [invoice_validation, payment_approval]
    permissions: [crm.read, payment.execute]

# LLM Gateway Integration
llm_gateway:
  api_base: "https://api.4sapi.com/v1"
  api_key: "${API_KEY}"
  models: [gpt-5.5, claude-4.6, gemini-3.1]