Enterprise AI Agent Deployment: Complete 3-Step Practical Framework

# Template: <system> <action> --param1=value --param2=value
# Examples
bpm approve --instanceId=xxx --user=zhangsan --comment="Compliant approval"
mail scan --today --tag=contract_approval
feishu notify --chatId=group_001 --msg="Approval 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):
        self.handlers[system] = handler

    def execute(self, command: str, user_id: str, **kwargs) -> Dict:
        # 1. Identity verification
        identity = self.auth_proxy.verify(user_id)
        # 2. Permission check
        if not self.permission_control.check(command, identity):
            return {"error": "permission_denied"}
        # 3. Rate limiting
        if not self.rate_limiter.allow(command):
            return {"error": "rate_limit_exceeded"}
        # 4. Dispatch command
        result = self._dispatch(command, kwargs)
        # 5. Audit log
        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:
        return self._call("bpm.approve", {"instance_id": instance_id, "user": user, "comment": comment})

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

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

from typing import Optional, List

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

    def _call(self, action: str, params: Dict) -> Dict:
        # Invoke 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:
        pass

    @abstractmethod
    def execute(self, context: Dict) -> Dict:
        pass

    def handle_exception(self, error: Exception, context: Dict) -> Dict:
        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 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:
        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. Risk screening via LLM
            risk_result = self._analyze_risk(context)
            if risk_result["level"] == "high":
                self._route_to_human_review(context)
                return {"status": "pending_review", "reason": "high_risk_contract"}

            # 2. Automated approval
            self.cli.execute("bpm approve", **risk_result["params"])
            self.cli.execute("feishu notify", **risk_result["notify_params"])
            return {"status": "completed", "decision": risk_result}

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

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

    def llm_call(self, prompt: str, model: str) -> str:
        # LLM call via 4sapi integration
        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:
        # 1. Detect event type
        event_type = self._detect_event(event)
        # 2. Check relevance
        if not self._is_relevant(event_type):
            return {"status": "ignored"}
        # 3. Match skill
        skill = self.skills.find(event_type, self.context.bound_skills)
        if not skill:
            return {"status": "no_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 Config
llm_gateway:
  api_base: "https://4sapi.com/v1"
  api_key: "${API_KEY}"
  models: [gpt-5.5, claude-4.6, gemini-3.1]