网关:把 agent 接进来

到第 13 章为止,jackdaw 会读配置(第 11 章)、会调上游(第 12 章)、会在路由组里挑模型(第 13 章)——但它只跟终端里的你说话。没有对外的门,每个代理(agent)都得自己管 API key,换模型要挨个改配置。这一章用 axum 把门装上:一个网关,对外只说 OpenAI 兼容的 chat 协议。

本章结束时,你应该能回答:

• chat_completions 的参数表里,body: Bytes 为什么必须排在最后,这些字节是谁交到它手上的;
• 一条 "stream": true 的请求,首块已经发给 agent,这时上游断了——网关为什么不再故障转移,而是补一条 SSE error 事件;
• 把 pi 指到 jackdaw 要填哪两个值,分别对应配置里的什么。

骨架:两条路由,一份状态

/// Shared per-process state: one router, one resolved token.
pub struct AppState {
    pub router: Arc<Router>,
    /// The token agents must present. Resolved once at startup so a missing
    /// env var fails fast (exit code 2) instead of 401-ing every request.
    pub token: String,
}

/// Assemble the axum app. Public so integration tests can serve it on a
/// random port with a known token.
pub fn build_app(state: Arc<AppState>) -> axum::Router {
    axum::Router::new()
        .route("/v1/chat/completions", post(chat_completions))
        .route("/v1/models", get(list_models))
        .with_state(state)
}

读法:从 build_app 看起,route 把"方法 + 路径"绑到 handler,整个网关只有两条路由。with_state 把一份 Arc<AppState> 挂进路由树,每个 handler 取回的都是同一份——Arc 是引用计数智能指针,克隆只加计数,不复制数据。状态只有两个字段:第 13 章的 Router,和启动时就解析好的 token。注释写明了为什么不在请求时才读:环境变量缺失应该在启动时以退出码 2 爆掉(第 10 章),而不是让每个请求 401 一遍。build_app 是 pub 的理由也在注释里:集成测试要在随机端口上把它跑起来(第 15 章展开)。serve 只绑定 127.0.0.1,端口取自配置的 gateway.port(启动时可用 --port 覆盖),默认 3426。

一条请求的字节级旅程

先把这条交互立为主线,后文都在追它的字节:

$ jackdaw serve &
$ curl -s 127.0.0.1:3426/v1/chat/completions -i \
    -H "Authorization: Bearer $JACKDAW_TOKEN" \
    -d '{"model":"group/coder","messages":[{"role":"user","content":"hi"}]}'
HTTP/1.1 200 OK
x-jackdaw-model: openai/gpt-5.2
{"choices":[{"message":{"role":"assistant","content":"Hello!"}}],...}

字节从 curl 出发:TCP 连上 127.0.0.1:3426,请求行、Authorization 头、JSON body 依次上线;axum 收完请求,把 body 的原始字节不做任何解析地交给 handler。

/// Order of extractors matters: the body (`Bytes`) must come last.
async fn chat_completions(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if !authorized(&headers, &state.token) {
        tracing::warn!("rejected a request without a valid bearer token");
        return error_response(
            StatusCode::UNAUTHORIZED,
            "Missing or invalid Authorization header. Send `Authorization: Bearer <token>` \
             where <token> is the value of the gateway's auth_token_env variable.",
        );
    }

    let request = match decode_request(&body) {
        Ok(request) => request,
        Err(message) => return error_response(StatusCode::BAD_REQUEST, message),
    };

    let requested = request.model_field().to_string();
    if request.wants_stream() {
        stream_reply(&state, request, &requested).await
    } else {
        complete_reply(&state, request, &requested).await
    }
}

读法:先看参数表。axum 的约定是:参数声明你要什么,axum 就从请求里取什么,这类参数叫提取器(extractor)。三个参数各取一份——共享状态、全部请求头、body 字节。注释点破顺序规矩:消耗 body 的提取器必须垫底,因为 body 只能被拿走一次,所有权的规矩在 HTTP 层照常生效(第 5 章)。函数体是三道关:authorized 验头,不过则 401;decode_request 用 serde_json::from_slice 从字节直接解析成 ChatRequest(第 12 章的 Value newtype),只额外要求"是 JSON 对象",否则 400;model_field().to_string() 把 "group/coder" 复制成 requested——时机关键,它在 request 被 move 进 router 之前留底,供日志与错误信息使用。最后按 wants_stream()(即 body 里的 "stream" 字段)选回程。

/// Bearer check: exact `Authorization: Bearer <token>` match.
fn authorized(headers: &HeaderMap, token: &str) -> bool {
    let Some(value) = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
    else {
        return false;
    };
    value
        .strip_prefix("Bearer ")
        .is_some_and(|presented| presented == token)
}

读法:let Some(value) … else 一眼看两头——头缺失,或值读不成文本(如含非 ASCII 字节),都直接 false;否则剥掉 Bearer 前缀,与启动时读到的 token 做一次明文的精确比较。没有哈希、没有常数时间比较;配合只监听 127.0.0.1,它防的是本机进程,不是公网。

接下来字节交棒给第 13 章的 router:解析 group/coder、按 order 策略选起点、把 body 里的 model 字段改写成 openai/gpt-5.2(其余原样),交给第 12 章的供应商发出。router 若报错,route_error_response 负责翻译:引用解析不了 → 400;上游判了 Fatal 的 4xx → 状态码与 body 原样透传,换模型无益;整组失败 → 502,details 里每个成员一行。第一次读记住这三层映射就够。

回程:整块 JSON、SSE 透传与 x-jackdaw-model

回程有两条。stream 不为 true 时,complete_reply 把 router 交回的 (used, response) 包成 (StatusCode::OK, Json(response.0)) 再盖章——curl 输出里 200 OK 和 x-jackdaw-model: openai/gpt-5.2 两行就是它的产物。stream 为 true 时走流式(SSE)那条路,多一层心思:

async fn stream_reply(state: &AppState, request: ChatRequest, requested: &str) -> Response {
    match state.router.stream(request).await {
        Ok((used, stream)) => {
            // ...
            let passthrough = stream.map(|item| match item {
                // Both arms yield `Ok`: a converted failure is data now,
                // not an error, so the body stream itself never fails.
                Ok(chunk) => Ok::<_, std::convert::Infallible>(chunk),
                Err(error) => Ok(sse_error_event(&error)),
            });
            let response = (
                StatusCode::OK,
                [
                    (header::CONTENT_TYPE, "text/event-stream"),
                    (header::CACHE_CONTROL, "no-cache"),
                ],
                Body::from_stream(passthrough),
            )
                .into_response();
            tag_model(response, &used)
        }
        // ...
    }
}

读法:从 stream.map 看起。stream 是 router 给的上游流(第 12 章的 SSE 逐块读取),map 对每一块做匹配:正常块原样放行——这就是透传,网关不拆解每个 data: 事件;出错的块被 sse_error_event 转成一条携带错误 JSON 的 SSE 事件。两个分支都返回 Ok,注释说明了用意:失败被转换成数据,这条流自身永不失败。Infallible 是"永不发生"的错误类型,把这句话写进了类型;Body::from_stream 把流变成响应体,配 text/event-stream 与 no-cache 两个头。

为什么不干脆让 Err 冒出去、或者中途换模型?因为首字节一旦离开进程,故障转移就不再可能——agent 会把两个模型的输出缝成一段怪答案。第 13 章定过这条语义,网关一侧的落实是:转移全部发生在 router.stream() 返回之前;之后再坏,就发一条 error 事件、诚实收尾。

透明哲学落在最后一处:

/// Attach the transparency header naming the model that actually answered.
fn tag_model(mut response: Response, used: &str) -> Response {
    // Model ids are config-controlled; a non-ASCII id would make an invalid
    // header, so the tag is skipped rather than dooming the whole response.
    if let Ok(value) = axum::http::HeaderValue::from_str(used) {
        response.headers_mut().insert("x-jackdaw-model", value);
    }
    response
}

读法:业务只有一行,把 used 塞进 x-jackdaw-model。if let Ok 的原因在注释里:模型名来自你的配置,若含非 ASCII 字符会造出非法 HTTP 头,这时宁可丢头,不让整个响应失败。网关不假装自己是某个模型,每次都报出实际应答者——透明在这里是一条可验证的响应头。

GET /v1/models:把组当模型卖

agent 的配置里要填 model id,而 jackdaw 出售的 id 只有组:

/// Advertise every group as a model id, in OpenAI's list format. Group ids
/// are the stable names agents configure once; members may change freely.
async fn list_models(State(state): State<Arc<AppState>>) -> Response {
    let data: Vec<Value> = state
        .router
        .config()
        .groups
        .keys()
        .map(|id| {
            json!({
                "id": format!("group/{id}"),
                // ...
            })
        })
        .collect();
    Json(json!({ "object": "list", "data": data })).into_response()
}

读法:从 state.router.config().groups.keys() 看起,配置里每张 [groups.X] 表的表名变成一个 group/X,套进 OpenAI 的 list 格式返回。中间那截 .map(…).collect() 是一条迭代器管道:map 拿闭包把每个键变形、collect 聚合回 Vec——本书只在用到处内联讲这两样(上文 stream_reply 里的 stream.map 是同一招),词条见术语表。只卖组的理由写在注释里:组名是 agent 配好就不动的稳定名字,而成员、顺序、策略可以随便改(第 13 章),agent 无感。注意这个 handler 没走 authorized——它只公布组名,不碰秘密。

把 Claude Code、pi 指到 jackdaw

现在回答开头第三问。对任何讲 OpenAI 协议的 agent,jackdaw 就是一个供应商,接入只填两个值:

• base URL:http://127.0.0.1:3426/v1,端口即第 11 章配置的 gateway.port;
• API key:$JACKDAW_TOKEN 的值——也就是 auth_token_env 指向的那个环境变量的值。它不会被送往任何上游,只用来过 authorized 这一关。

对 pi:它的模型与供应商配置放在 ~/.pi/ai.json,且支持自定义供应商(custom provider,baseUrl + apiKey)——按上面两行填,jackdaw 就是 pi 眼里的一个供应商。对 Claude Code:同样是改 base URL 和 API key 两处;具体字段名以它当下的文档为准,这里不断言。

TIP

配完先自检:curl -s 127.0.0.1:3426/v1/models 应列出你的全部 group/<id>,再用上面那条 curl 验证拿到 200 与 x-jackdaw-model。

对照 magpie:它会把你选中的模型写进各 agent 的配置文件;jackdaw 刻意不做——手工两处,换你亲眼看见 agent 在读什么。这是"magpie 有、jackdaw 不做"清单上的一项,是教学点,不是缺陷。

小结

• 网关职责刻意压窄:验 token、验 JSON、交给 router、翻译回 HTTP;上游语义都在第 13 章的 router 里。
• 提取器按序取请求,消耗 body 的 Bytes 必须垫底;requested 在 request 被 move 走之前复制留底。
• 流式透传:错误转成 SSE 事件,流自身永不失败;首字节之后不再转移。
• x-jackdaw-model 每次应答都报出实际使用的模型。
• 下一章(第 15 章)给这套网关上保险:本地 stub provider、集成测试与 CI。

自检

1. 迁移:你想让网关同时接受 Authorization: Bearer … 与 x-api-key: … 两种认证,该改哪个函数?返回值的语义要变吗?
2. 预测输出:token 正确,但 body 是合法 JSON 的数组 [1,2,3]——状态码是多少?错误信息是哪一句?
3. 解释:stream_reply 为什么把上游错误转成 Ok(sse_error_event(...)),而不是让 Err 直接冒出去?

参考答案

1. 改 authorized。它现在只读 Authorization 头;要支持第二种,就同时尝试 x-api-key,两个头都没给出正确值才返回 false。返回值语义不变——仍是"这次请求放不放行"——变的只是判定逻辑;chat_completions 里的 401 提示文案也应把两种发法都写上,保持可行动。
2. 400。走 decode_request 的第二条岔路:value.is_object() 不成立,返回的正是那句 "request body must be a JSON object"。
3. 响应体一旦开始发送,就再没有机会改状态码或发一个普通的错误响应;让 Err 冒出去等于流半途无声断掉。转成一条 data: 事件,失败原因在协议内送达,然后流正常收尾——诚实失败。