def check_auth_status() -> dict:
    status = {
        'api_key_set': bool(os.environ.get('ANTHROPIC_API_KEY')),
        'oauth_logged_in': False,
    }

    # 檢查 OAuth 登入狀態
    claude_dir = os.path.expanduser('~/.claude')
    if os.path.exists(claude_dir):
        auth_files = ['credentials.json', 'settings.json', '.credentials.json']
        for auth_file in auth_files:
            if os.path.exists(os.path.join(claude_dir, auth_file)):
                status['oauth_logged_in'] = True
                break

    # 只需要其中一種認證
    status['auth_available'] = status['api_key_set'] or status['oauth_logged_in']
    return status

try:
    from claude_code_sdk._internal.transport import subprocess_cli
    subprocess_cli._MAX_BUFFER_SIZE = 50 * 1024 * 1024  # 50MB
except Exception as e:
    print(f"Failed to patch SDK buffer size: {e}")

_claude_service: Optional['ClaudeCodeService'] = None

def get_claude_code() -> 'ClaudeCodeService':
    global _claude_service
    if _claude_service is None:
        _claude_service = ClaudeCodeService()
    return _claude_service

@dataclass
class AgentConfig:
    allowed_tools: List[str]
    max_turns: int = 30
    cwd: Optional[str] = None

    @classmethod
    def coding(cls) -> 'AgentConfig':
        """開發模式：完整權限"""
        return cls(
            allowed_tools=['Read', 'Write', 'Edit', 'Bash', 'Glob', 'Grep'],
            max_turns=50
        )

    @classmethod
    def research(cls) -> 'AgentConfig':
        """研究模式：可上網"""
        return cls(
            allowed_tools=['Read', 'Glob', 'Grep', 'WebFetch'],
            max_turns=20
        )

    @classmethod
    def readonly(cls) -> 'AgentConfig':
        """唯讀模式：最安全"""
        return cls(
            allowed_tools=['Read', 'Glob', 'Grep'],
            max_turns=10
        )

undefined

undefined

undefined

undefined

from dataclasses import dataclass, field
from typing import AsyncIterator, Optional, List, Dict, Any
from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock

@dataclass
class ClaudeCodeService:
    """Claude Code 服務封裝"""

    async def run(self, prompt: str, config: Optional[AgentConfig] = None) -> Dict[str, Any]:
        """執行並收集完整結果"""
        options = ClaudeCodeOptions(
            allowed_tools=config.allowed_tools if config else None,
            max_turns=config.max_turns if config else 30,
            cwd=config.cwd
        )

        result_texts = []
        tool_calls = []

        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        result_texts.append(block.text)

        return {
            'status': 'success',
            'result': '\n'.join(result_texts),
            'tool_calls': tool_calls
        }

    async def stream(self, prompt: str, config: Optional[AgentConfig] = None) -> AsyncIterator[dict]:
        """串流執行，即時回傳"""
        options = ClaudeCodeOptions(
            allowed_tools=config.allowed_tools if config else None,
            max_turns=config.max_turns if config else 30
        )

        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        yield {'type': 'text', 'content': block.text}

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/api/agent/stream")
async def agent_stream(prompt: str):
    async def event_generator():
        service = ClaudeCodeService()
        async for msg in service.stream(prompt):
            yield f"data: {json.dumps(msg)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

undefined

undefined

undefined

options = ClaudeCodeOptions(max_turns=request.max_turns)
results = []

async for message in query(prompt=request.prompt, options=options):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                results.append(block.text)

return {"result": "\n".join(results)}

async def event_generator():
    options = ClaudeCodeOptions(max_turns=request.max_turns)
    async for message in query(prompt=request.prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    yield f"data: {json.dumps({'text': block.text})}\n\n"
    yield "data: [DONE]\n\n"

return StreamingResponse(event_generator(), media_type="text/event-stream")

**requirements.txt**:

**Explanation**: 這個後端會自動檢測 OAuth 或 API Key 認證，支援一般請求和串流回應。

// 一般請求
async function generate(prompt) {
    const response = await fetch('http://localhost:8000/api/generate', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });
    const data = await response.json();
    return data.result;
}

// 串流請求
async function streamGenerate(prompt, onChunk) {
    const response = await fetch('http://localhost:8000/api/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');

        for (const line of lines) {
            if (line.startsWith('data: ') && line !== 'data: [DONE]') {
                const data = JSON.parse(line.slice(6));
                onChunk(data.text);
            }
        }
    }
}

undefined

```yaml

undefined