claude-code-acp-py（下）：ACP Proxy 橋接多 AI 後端

# Copilot SDK 相容 flags（接受但不使用）
parser.add_argument("--headless", action="store_true")
parser.add_argument("--server", action="store_true")
parser.add_argument("--stdio", action="store_true")
parser.add_argument("--auth-token-env", default="")
parser.add_argument("--no-auto-login", action="store_true")

# Proxy 自己的 flags
parser.add_argument("--backend", default=os.environ.get("ACP_PROXY_BACKEND", "gemini"))
parser.add_argument("--backend-args", nargs="*", default=[])
parser.add_argument("--cwd", default=os.getcwd())

handlers = {
    "ping": self._handle_ping,
    "status.get": self._handle_status_get,
    "auth.getStatus": self._handle_auth_get_status,
    "models.list": self._handle_models_list,
    "session.create": self._handle_session_create,
    "session.resume": self._handle_session_resume,
    "session.send": self._handle_session_send,
    "session.destroy": self._handle_session_destroy,
    "session.abort": self._handle_session_abort,
    "session.list": self._handle_session_list,
    "session.delete": self._handle_session_delete,
    "session.getMessages": self._handle_session_get_messages,
    "session.getLastId": self._handle_session_get_last_id,
    # ...
}

if self.backend == "gemini":
    models = [
        {"id": "gemini-2.0-flash", "name": "Gemini 2.0 Flash", ...},
        {"id": "gemini-1.5-pro", "name": "Gemini 1.5 Pro", ...},
    ]
elif self.backend in ("claude-code", "claude-code-acp"):
    models = [
        {"id": "claude-sonnet-4-20250514", "name": "Claude Sonnet 4", ...},
        {"id": "claude-opus-4-20250514", "name": "Claude Opus 4", ...},
    ]

class SessionEventType(str, Enum):
    SESSION_START = "session.start"
    SESSION_RESUME = "session.resume"
    SESSION_IDLE = "session.idle"
    ASSISTANT_TURN_START = "assistant.turn_start"
    ASSISTANT_MESSAGE = "assistant.message"
    ASSISTANT_MESSAGE_DELTA = "assistant.message_delta"
    ASSISTANT_REASONING_DELTA = "assistant.reasoning_delta"
    TOOL_EXECUTION_START = "tool.execution_start"
    TOOL_EXECUTION_COMPLETE = "tool.execution_complete"
    # ...

# 依後端類型決定 CLI 參數
if self.backend_command == "gemini":
    if "--experimental-acp" not in backend_args:
        backend_args.append("--experimental-acp")
    # Gemini: 用 CLI 參數傳 model
    if model:
        backend_args.extend(["--model", model])

elif self.backend_command in ("claude", "claude-code", "claude-code-acp"):
    # Claude: 用 ACP set_session_model 方法傳 model
    pass

elif self.backend_command == "copilot":
    if "--acp" not in backend_args:
        backend_args.append("--acp")
    # Copilot: 也用 CLI 參數
    if model:
        backend_args.extend(["--model", model])

@client.on_text
async def on_text(text: str):
    event = create_assistant_message_delta_event(text)
    session.events.append(event)
    if session.event_callback:
        await session.event_callback(event)

@client.on_tool_start
async def on_tool_start(tool_id: str, name: str, input_data: dict):
    event = create_tool_execution_start_event(tool_id, name, input_data)
    if session.event_callback:
        await session.event_callback(event)

def _build_models_state(self, server_info):
    models_data = server_info.get("models", [])
    available_models = []
    for model in models_data:
        available_models.append(
            ModelInfo(
                model_id=model.get("value", ""),
                name=model.get("displayName", model_id),
            )
        )
    return SessionModelState(
        available_models=available_models,
        current_model_id=current_model_id,
    )

async def _send_available_commands(self, session_id, server_info):
    commands_data = server_info.get("commands", [])
    # 過濾不適合 ACP 的指令
    unsupported = {"cost", "login", "logout", "release-notes", ...}

    available_commands = []
    for cmd in commands_data:
        if cmd["name"] not in unsupported:
            available_commands.append(
                AvailableCommand(name=name, description=cmd.get("description", ""))
            )

    await self._conn.session_update(
        session_id,
        update_available_commands(available_commands),
    )

# Copilot 格式
{"nanobanana": {"type": "local", "command": "uvx", "args": ["nanobanana"], "tools": ["*"]}}

# ACP 格式
[{"name": "nanobanana", "command": "uvx", "args": ["nanobanana"]}]

if isinstance(v, str) and v.startswith("${") and v.endswith("}"):
    var_name = v[2:-1]
    env[k] = os.environ.get(var_name, "")

def test_client_initialization_custom(self):
    client = AcpClient(
        command="my-agent",
        args=["--verbose"],
        cwd="/tmp",
        env={"KEY": "value"},
    )
    assert client.command == "my-agent"
    assert client.args == ["--verbose"]

def test_on_text_decorator(self):
    client = AcpClient()

    @client.on_text
    async def handler(text: str):
        pass

    assert client.events.on_text is handler

requires_claude_cli = pytest.mark.skipif(
    not is_claude_cli_available(),
    reason="Claude CLI not available",
)

import asyncio
import os
from copilot import CopilotClient

os.environ["ACP_PROXY_BACKEND"] = "gemini"

async def main():
    client = CopilotClient({"cli_path": "copilot-acp-proxy"})
    await client.start()

    session = await client.create_session({"model": "gemini-2.0-flash"})

    def on_event(event):
        event_type = event.type.value if hasattr(event.type, 'value') else str(event.type)
        if event_type == "assistant.message_delta":
            delta = getattr(event.data, 'deltaContent', None)
            if delta:
                print(delta, end="", flush=True)

    session.on(on_event)
    await session.send({"prompt": "Hello from Gemini via Proxy!"})

asyncio.run(main())

session = await client.create_session({"model": "opus"})