跳转至内容

将 GitHub Copilot SDK 集成到应用中

本演练讲解如何将 FastAPI 应用与 GitHub Copilot SDK 集成,从而将 Copilot 会话转换为应用程序服务。您将了解该应用如何启动 SDK 客户端、创建流式传输会话、监听会话事件、将回调桥接到异步生成器,并避免使用过期的模型目录。

SDK 使用位置

主要集成点为 app/services/copilot_chat.py。其通过以下方式导入 SDK:

from copilot import CopilotClient, SessionEvent, SessionEventType

PyPI 包名为 github-copilot-sdk,在 pyproject.toml 中固定版本为 1.0.9,但导入根目录为 copilot

requires-python = ">=3.11"
dependencies = [
    "github-copilot-sdk==1.0.9",

需要 Python 3.11 或更高版本。

一个长生命周期服务

app/main.py 为 FastAPI 应用的生命周期创建一个聊天服务:

app.state.chat_service = CopilotChatService()

该服务按需连接。ensure_started() 由一个 asyncio.Lock 保护,因此并发 HTTP 请求不会同时启动两个传输:

self._lock = asyncio.Lock()
async with self._lock:
    if self._is_started and self._client is not None:
        return

启动和关闭是显式的:

self._client = CopilotClient()
await self._client.start()
self._is_started = True
if self._client is not None:
    await self._client.stop()

SDK 还支持 async with CopilotClient() 用于短生命周期脚本;实验室示例在客户端仅持续一个命令时使用该形式。

创建流式会话

CopilotChatService.chat_stream() 为每个提示创建一个新的 SDK 会话:

session = await self._client.create_session(
    model=model,
    streaming=True,
    system_message=(
        {"mode": "append", "content": system_message} if system_message else None
    ),
)

create_session(...) 仅接受关键字参数。重要参数包括 modelstreamingsystem_messagetoolsmcp_serverssession_idon_permission_request

system_message 是一个 TypedDict 联合类型:

{"mode": "append", "content": "..."}
{"mode": "replace", "content": "..."}

演示使用追加模式,因此应用程序上下文在不覆盖基础 Copilot 行为的情况下被添加。

⚠️ Python 自定义工具需要 on_permission_request 或者调用会被拒绝。.NET 示例无需处理相同简单的工具流程。Python 在使用来自 copilot.rpc 的权限决策时,也无需通过 GHCP001 抑制实验性 API 警告。

会话事件

核心架构要点:Python SDK 的事件为 仅推送回调(push-only callbacks)session.on(handler) 注册处理器并返回取消订阅的可调用对象;不存在异步迭代器。

与 C# SDK 不同,Python 暴露了 一个 SessionEvent 数据类。您通过判断 evt.type(一个 SessionEventType 枚举值)来分支处理,而非为每个事件类型匹配一个子类。此为实际处理器:

def on_event(evt: SessionEvent) -> None:
    # Unlike .NET, every event arrives as one SessionEvent
    # carrying a `type` enum and a `data` payload, so this
    # dispatches on `evt.type` rather than on subclasses.
    if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
        queue.put_nowait(evt.data.delta_content or "")
    elif evt.type is SessionEventType.ASSISTANT_MESSAGE:
        logger.info(
            "Assistant response complete: %d chars",
            len(evt.data.content or ""),
        )
    elif evt.type is SessionEventType.SESSION_IDLE:
        if not done.done():
            done.set_result(None)
    elif evt.type is SessionEventType.SESSION_ERROR:
        logger.error("Session error: %s", evt.data.message)
        if not done.done():
            done.set_exception(RuntimeError(evt.data.message))

ASSISTANT_MESSAGE_DELTA 传输流式文本,ASSISTANT_MESSAGE 标记完整回答,SESSION_IDLE 表示当前轮次的等待结束,SESSION_ERROR 则表示失败。

将回调桥接到异步生成器

FastAPI 流式传输需要异步迭代器,但 SDK 调用处理器。服务通过将回调桥接到 asyncio.Queue 并从 chat_stream() 中抽取数据来实现:

queue: asyncio.Queue[object] = asyncio.Queue()
loop = asyncio.get_running_loop()
item = await queue.get()
if item is _DONE:
    break
if isinstance(item, BaseException):
    raise item
yield item  # type: ignore[misc]

_DONE 是一个哨兵对象,而非内容值:

# Sentinel pushed onto the queue when the session goes idle.
_DONE = object()
queue.put_nowait(_DONE)

由于事件回调无法直接向 HTTP 响应 yield 数据,因此需要使用队列。这与 C# 服务将数据写入 System.Threading.Channels 通道的行为完全一致。

重连行为

若 SDK 传输丢失,服务将客户端标记为不健康,并将异常传递至队列:

except (ConnectionError, OSError) as ex:
    logger.warning("Copilot connection lost: %s", ex)
    self._is_started = False
    queue.put_nowait(ex)

当前请求失败,下一请求会再次调用 ensure_started()。由于 _is_started 为 false,旧客户端将被停止,新传输通道将被建立。

列出实时模型

CopilotChatService.list_models() 会向已连接的 Copilot CLI 查询当前已登录账户可使用的模型:

models = await self._client.list_models()
return [(m.id, m.name or m.id) for m in models or [] if m.id]

SDK 返回包含 ModelInfo 对象的 .id.name 字段。硬编码模型 ID 存在风险,因为模型可用性会因账户和发布计划而变化。

已知上游问题:调用 list_models() 可能抛出 ValueError: Missing required field 'multiplier' in ModelBilling(详见 github/copilot-sdk#1302)。app/routers/chat.py 会捕获异常并回退到包含六个模型的静态目录:

except Exception:
    # Includes the known SDK issue where ModelBilling is missing the
    # required 'multiplier' field: github/copilot-sdk#1302.
    logger.warning(
        "Could not list models from Copilot CLI; using static catalog", exc_info=True
    )

return list(AVAILABLE_MODELS.values())

这确保了即使实时发现功能暂时中断,模型选择器仍可正常使用。