Claude 命令行工具开发实战：从原理到高效实现

1次阅读

没有评论

共计 3175 个字符，预计需要花费 8 分钟才能阅读完成。

Claude API 提供了强大的自然语言处理能力，但直接调用 API 时开发者常会遇到以下问题：

认证流程繁琐 ：每次请求都需要处理 API key 和认证头信息
响应解析复杂 ：流式响应需要特殊处理，JSON 结构嵌套较深
缺乏交互性 ：直接调用 API 无法实现类似命令行的交互体验
调试困难 ：错误处理和日志记录需要额外开发

我们比较了几种主流语言实现命令行工具的优缺点：

Python：
优点：丰富的库支持 (requests, argparse 等)，开发速度快
缺点：性能相对较低，打包部署稍复杂
Go：
优点：高性能，静态编译方便部署
缺点：生态相对 Python 略少，开发效率稍低
Rust：
优点：极致性能，内存安全
缺点：学习曲线陡峭，开发周期长

推荐选择 ：对于大多数场景，Python 是最佳平衡点，特别是需要快速迭代时。

import os
from pathlib import Path

def load_api_key():
    """从环境变量或配置文件中加载 API key"""
    # 优先级 1：环境变量
    key = os.getenv('CLAUDE_API_KEY')
    if key:
        return key

    # 优先级 2：配置文件
    config_path = Path.home() / '.claude_config'
    if config_path.exists():
        with open(config_path, 'r') as f:
            return f.read().strip()

    raise ValueError("API key not found in environment or config file")

import requests

def send_request(prompt, model="claude-2", max_tokens=1000):
    url = "https://api.anthropic.com/v1/complete"
    headers = {
        "Content-Type": "application/json",
        "X-API-Key": load_api_key()}

    payload = {"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
        "model": model,
        "max_tokens_to_sample": max_tokens,
        "stream": True
    }

    response = requests.post(url, headers=headers, json=payload, stream=True)
    response.raise_for_status()
    return response

def handle_stream_response(response):
    """处理流式响应，实时输出结果"""
    buffer = ""
    for line in response.iter_lines():
        if line:
            decoded_line = line.decode('utf-8')
            if decoded_line.startswith('data:'):
                try:
                    data = json.loads(decoded_line[5:])
                    if 'completion' in data:
                        print(data['completion'], end='', flush=True)
                except json.JSONDecodeError:
                    continue

def safe_execute(prompt):
    try:
        response = send_request(prompt)
        handle_stream_response(response)
    except requests.exceptions.HTTPError as e:
        print(f"API 请求失败: {e.response.status_code}")
        if e.response.status_code == 429:
            print("请求过于频繁，请稍后再试")
    except ValueError as e:
        print(f"配置错误: {e}")
    except Exception as e:
        print(f"未知错误: {e}")

#!/usr/bin/env python3
import argparse
import json
import os
import requests
from pathlib import Path

# ... 前面定义的函数 ...

def main():
    parser = argparse.ArgumentParser(description='Claude 命令行工具')
    parser.add_argument('prompt', help='输入提示词')
    parser.add_argument('--model', default='claude-2', help='模型版本')
    parser.add_argument('--max-tokens', type=int, default=1000, help='最大 token 数')

    args = parser.parse_args()
    safe_execute(args.prompt, args.model, args.max_tokens)

if __name__ == "__main__":
    main()

并发请求 ：使用 aiohttp 实现异步请求

import aiohttp
import asyncio

async def async_send_request(session, prompt):
    url = "https://api.anthropic.com/v1/complete"
    headers = {
        "Content-Type": "application/json",
        "X-API-Key": load_api_key()}

    payload = {"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
        "model": "claude-2",
        "max_tokens_to_sample": 1000,
        "stream": False
    }

    async with session.post(url, headers=headers, json=payload) as response:
        return await response.json()

缓存策略 ：对常见查询结果进行缓存

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_request(prompt):
    # 实现带缓存的请求
    return send_request(prompt)

永远不要将 API key 硬编码在代码中
使用环境变量或加密的配置文件存储敏感信息
实现请求限流，避免触发 API 限制

from ratelimit import limits, sleep_and_retry

# 限制每分钟 50 次请求
@sleep_and_retry
@limits(calls=50, period=60)
def rate_limited_request(prompt):
    return send_request(prompt)