两个数字员工，一个组织：企业数字员工开发最佳实践

import subprocess
import json
from dataclasses import dataclass


@dataclass
class TaskRequest:
    """从钉钉消息解析出的任务请求。"""
    title: str
    body: str
    labels: list[str]
    sender_id: str
    conversation_id: str


def create_issue(task: TaskRequest) -> int:
    """创建 GitHub Issue 并返回编号。"""
    result = subprocess.run(
        [
            "gh", "issue", "create",
            "--title", task.title,
            "--body", task.body,
            "--label", ",".join(task.labels),
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    # gh issue create 输出格式: https://github.com/.../issues/12
    issue_number = int(result.stdout.strip().rsplit("/", 1)[-1])
    return issue_number
# generated by hugo AI

dingtalk-opencode-tag/
├── core/                  # 基础设施（不动）
│   ├── dws_consumer.py    # dws consume 长连接
│   ├── message_router.py  # 消息路由
│   └── state_manager.py   # 状态管理
├── custom/                # 业务能力插件（改这里）
│   ├── calendar/
│   │   ├── freebusy.py    # ← 新增：闲忙查询
│   │   └── test_freebusy.py
│   ├── todo/
│   ├── approval/
│   └── ...
├── Dockerfile
├── docker-compose.yml
└── .github/workflows/deploy.yml

import subprocess
from pathlib import Path


def commit_and_push(issue_number: int, message: str) -> None:
    """提交代码并推送，触发 CI/CD。"""
    repo = Path("~/Projects/dingtalk-opencode-tag").expanduser()

    subprocess.run(["git", "add", "-A"], cwd=repo, check=True)
    subprocess.run(
        ["git", "commit", "-m", f"feat(calendar): 闲忙查询 Closes #{issue_number}"],
        cwd=repo,
        check=True,
    )
    subprocess.run(
        ["git", "push", "origin", "main"],
        cwd=repo,
        check=True,
        env={
            **__import__("os").environ,
            "GIT_SSH_COMMAND": "ssh -i ~/.ssh/id_ed25519",
        },
    )
# generated by hugo AI

# .github/workflows/deploy.yml
name: Build & Deploy Digital Employee

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.13"
      - run: pip install -r requirements.txt
      - run: python -m pytest custom/ -v

  build-and-deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build Docker image
        run: |
          docker build -t ${{ secrets.REGISTRY }}/digital-employee:${{ github.sha }} .

      - name: Push image
        run: |
          echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login -u ${{ secrets.REGISTRY_USER }} --password-stdin
          docker push ${{ secrets.REGISTRY }}/digital-employee:${{ github.sha }}

      - name: Deploy to server
        uses: appleboy/ssh-action@v1
        with:
          host: ${{ secrets.DEPLOY_HOST }}
          username: deploy
          key: ${{ secrets.DEPLOY_SSH_KEY }}
          script: |
            docker pull ${{ secrets.REGISTRY }}/digital-employee:${{ github.sha }}
            docker stop digital-employee || true
            docker rm digital-employee || true
            docker run -d \
              --name digital-employee \
              --restart=always \
              -e DWS_TOKEN=*** \
              -e OPENCODE_MODEL=*** \
              -e DINGTALK_CORP_ID=*** \
              -e SUPERVISOR_USER_ID=*** \
              -e IMAGE_TAG=${{ github.sha }} \
              ${{ secrets.REGISTRY }}/digital-employee:${{ github.sha }}
# generated by hugo AI

import os
import asyncio
import subprocess
import json
from datetime import datetime


async def report_startup() -> None:
    """数字员工上线后主动向主管打招呼。"""
    supervisor_id = os.environ["SUPERVISOR_USER_ID"]
    image_tag = os.environ.get("IMAGE_TAG", "unknown")
    start_time = datetime.now()

    # 自检：加载所有插件，检查状态
    plugins = load_plugins()
    plugin_status = []
    for name, plugin in plugins.items():
        try:
            await plugin.health_check()
            plugin_status.append(f"{name} ✅")
        except Exception as e:
            plugin_status.append(f"{name} ❌ ({e})")

    startup_seconds = (datetime.now() - start_time).total_seconds()

    message = (
        f"🤖 数字员工 {image_tag[:7]} 已上线\n"
        f"━━━━━━━━━━━━━━━━\n"
        f"状态：{'✅ 正常' if all('✅' in s for s in plugin_status) else '⚠️ 部分异常'}\n"
        f"插件：{' | '.join(plugin_status)}\n"
        f"启动耗时：{startup_seconds:.1f}s\n"
        f"━━━━━━━━━━━━━━━━\n"
        f"随时 @我 开始工作。"
    )

    # 通过 dws 发送单聊消息给主管
    subprocess.run(
        [
            "dws", "chat", "send",
            "--user", supervisor_id,
            "--content", message,
        ],
        check=True,
    )
# generated by hugo AI

用户在群里说「这个功能不好用」
       │
       ▼
我看到 → 在管理群 @Coding Agent「修一下 #15」
       │
       ▼
Coding Agent: dws event consume 收到
       │
       ├─ gh issue view 15（读取需求）
       ├─ 开发、测试
       ├─ git commit + push
       │
       ▼
GitHub CI/CD: build → test → docker → deploy
       │
       ▼
数字员工重启 → 自检 → 向我报告「✅ v1.3.2 已上线」
       │
       ▼
用户在群里 @数字员工，发现功能修好了
       │
       ▼
新的反馈 → 新的迭代