Skill 异常处理实践：让 AI Agent 告别静默失败

# ❌ 危险：失败时返回空列表，LLM 会误以为「没有需求」
def get_stories(workspace_id, iteration_id):
    result = call_safe('stories_get', {...})
    if not result['ok']:
        return []
    return extract_list(result['data'])

# ✅ 正确：失败时抛出异常，携带错误类型
def get_stories(workspace_id, iteration_id):
    result = call_safe('stories_get', {...})
    if not result['ok']:
        raise TapdApiError(
            f'获取需求列表失败: {result["error"]}',
            error_type=result.get('error_type', TapdApiError.UNKNOWN)
        )
    return extract_list(result['data'])

{
  "ok": false,
  "error_type": "auth_failed",
  "error": "TAPD 鉴权失败，请确认 mcporter-internal 已完成 tapd 授权配置",
  "workspace_id": "70132312",
  "llm_hint": "🔴 TAPD 鉴权失败，请确认 mcporter-internal 已完成 tapd 授权配置。"
}

class TapdApiError(Exception):
    # 错误类型常量
    TIMEOUT = 'timeout'
    NOT_FOUND = 'command_not_found'
    AUTH_FAILED = 'auth_failed'
    EMPTY_RESPONSE = 'empty_response'
    JSON_TRUNCATED = 'json_truncated'
    JSON_PARSE = 'json_parse'
    API_ERROR = 'api_error'
    PROCESS_ERROR = 'process_error'
    UNKNOWN = 'unknown'

    def __init__(self, message, error_type='unknown', raw=''):
        super().__init__(message)
        self.error_type = error_type
        self.raw = raw  # 原始输出，用于调试

# call_safe() 返回结构化结果，不抛出异常
def call_safe(tool_name, tool_args, timeout=30):
    try:
        data = call(tool_name, tool_args, timeout)
        return {'ok': True, 'data': data}
    except TapdApiError as e:
        return {'ok': False, 'error': str(e), 'error_type': e.error_type, 'data': []}
    except Exception as e:
        return {'ok': False, 'error': f'未预期的异常: {e}', 'error_type': TapdApiError.UNKNOWN, 'data': []}

def exit_with_error(error_type, message, workspace_id='', hint=''):
    output = {
        'ok': False,
        'error_type': error_type,
        'error': message,
        'workspace_id': workspace_id,
        'llm_hint': hint or f'🔴 脚本执行失败（{error_type}）：{message}',
    }
    print(json.dumps(output, ensure_ascii=False, indent=2))
    sys.exit(1)

# 在 main() 中使用
def main():
    try:
        stories = tapd_api.get_stories(workspace_id, iteration_id)
    except tapd_api.TapdApiError as e:
        exit_with_error(
            error_type=e.error_type,
            message=str(e),
            hint=_error_hint(e)  # 根据 error_type 生成针对性提示
        )

def get_stories(workspace_id, iteration_id, limit=200):
    PAGE_SIZE = 10  # 经验值，10 条约 2-3KB，安全边际充足
    all_items, page, first_page = [], 1, True

    while True:
        result = call_safe('stories_get', {**base_args, 'page': page})
        if not result['ok']:
            if first_page:
                raise TapdApiError(...)   # 第一页失败：致命错误
            else:
                print(f'⚠️ 第{page}页失败，停止分页', file=sys.stderr)
                break                      # 后续页失败：降级处理

        first_page = False
        page_items = extract_list(result['data'])
        if not page_items:
            break

        all_items.extend(page_items)
        page += 1
        if len(page_items) < PAGE_SIZE:
            break  # 最后一页
    return all_items

# 改造前：裸抛，LLM 拿到的是 traceback
with urllib.request.urlopen(req, timeout=15) as resp:
    data = json.loads(resp.read())
return data.get("html_url", "")
# HTTPError 直接往上飞，main() 没有 try/except

# 错误类型常量
ERR_GIT       = "git_error"
ERR_AUTH      = "auth_failed"
ERR_NETWORK   = "network_error"
ERR_EXISTS    = "already_exists"
ERR_NOT_FOUND = "not_found"

_HINT_MAP = {
    ERR_AUTH:    "🔴 GitHub 鉴权失败（401/403），请确认 git remote URL 中的 PAT 有效。",
    ERR_NETWORK: "🔴 网络请求失败，请检查内网/外网连通性，稍后重试。",
    ERR_EXISTS:  "🟡 PR 分支已存在，请检查 GitHub 上是否已有对应 PR。",
    # ...
}

def exit_with_error(error_type: str, message: str, detail: str = "") -> None:
    output = {
        "ok": False,
        "error_type": error_type,
        "error": message,
        "detail": detail,
        "llm_hint": _HINT_MAP.get(error_type, "🔴 未知错误"),
    }
    print(json.dumps(output, ensure_ascii=False, indent=2))
    sys.exit(1)

except urllib.error.HTTPError as e:
    if e.code in (401, 403):
        exit_with_error(ERR_AUTH, f"GitHub 鉴权失败（HTTP {e.code}）")
    elif e.code == 422:
        if "pull request already exists" in detail.lower():
            exit_with_error(ERR_EXISTS, "该分支已存在对应的 PR，请勿重复创建")
        exit_with_error(ERR_PR_CREATE, f"PR 创建失败（HTTP {e.code}）：参数错误")
    else:
        exit_with_error(ERR_PR_CREATE, f"PR 创建失败（HTTP {e.code}）")

# 前置校验：私钥文件存在（不等 SSH 进程崩溃再报错）
if not os.path.exists(SSH_KEY):
    exit_with_error(ERR_SSH_KEY, f"SSH 私钥文件不存在：{SSH_KEY}")

# 超时单独捕获
except subprocess.TimeoutExpired:
    exit_with_error(ERR_TIMEOUT, "SSH 命令超时（120s）")

# 根据退出码和 stderr 推断 SSH 错误类型
def _classify_ssh_error(rc: int, stderr: str) -> str:
    if rc == 255:
        if "connection refused" in stderr or "no route to host" in stderr:
            return ERR_SSH_CONNECT   # 连不上
        if "permission denied" in stderr or "publickey" in stderr:
            return ERR_SSH_AUTH      # 密钥问题
    return ERR_MAKE_FAILED           # make 本身失败

# 改造前：静默 fallback，用户不知道中文会乱码
return ImageFont.load_default()

# 改造后：输出 WARNING，降级继续（非致命），LLM/用户可感知
if not _font_missing_warned:
    print(
        "WARNING: 未找到 CJK 字体，中文标题将显示为方块。"
        " 请运行：dnf install google-noto-cjk-fonts",
        file=sys.stderr,
    )
    _font_missing_warned = True
return ImageFont.load_default()

{
  "ok": true,
  "path": "/projects/hxblog/.../banner.jpg",
  "warnings": ["CJK 字体缺失，中文标题可能显示为方块，请安装 google-noto-cjk-fonts"],
  "llm_hint": "✅ Banner 生成成功  ⚠️ 注意：CJK 字体缺失..."
}