futu_mcp/
http.rs

1use std::path::PathBuf;
2
3use anyhow::{Context, Result};
4
5use crate::tools;
6
7/// HTTP 模式：axum + rmcp StreamableHttpService，`/mcp` 路径跑 MCP，
8/// `/metrics` 暴露 Prometheus counters（无需 token），
9/// `/.well-known/oauth-protected-resource` 暴露 OAuth2 Protected Resource
10/// Metadata（RFC 9728，给 MCP 客户端发现鉴权要求用）。
11///
12/// v1.4+：未带 Bearer token 的 `/mcp` 请求会回 `401 + WWW-Authenticate`
13/// 头，指向 resource metadata，配合 rmcp 客户端的自动发现流程。
14pub(super) fn render_mcp_metrics_body() -> String {
15    let registry = futu_auth::metrics::global();
16    render_mcp_metrics_body_for(registry.as_deref())
17}
18
19pub(super) fn render_mcp_metrics_body_for(registry: Option<&futu_auth::MetricsRegistry>) -> String {
20    registry.map(|r| r.render_prometheus()).unwrap_or_else(|| {
21        concat!(
22            "# HELP futu_metrics_registry_installed Whether futu_auth metrics registry is installed (1=yes, 0=no)\n",
23            "# TYPE futu_metrics_registry_installed gauge\n",
24            "futu_metrics_registry_installed{state=\"metrics registry not installed\"} 0\n"
25        )
26        .to_string()
27    })
28}
29
30pub(super) async fn serve_http(
31    server: tools::FutuServer,
32    listen: &str,
33    tls: Option<(PathBuf, PathBuf)>,
34) -> Result<()> {
35    use rmcp::transport::streamable_http_server::{
36        StreamableHttpService, session::local::LocalSessionManager,
37    };
38
39    // 补齐 `:port` 写法：bind 到 0.0.0.0:port
40    let bind_addr = if listen.starts_with(':') {
41        format!("0.0.0.0{listen}")
42    } else {
43        listen.to_string()
44    };
45
46    // rmcp StreamableHttpService 要求一个 service factory —— 每个 HTTP 会话要
47    // 一份独立的 MCP Service 实例。FutuServer 目前是 Clone 的（ServerState 内
48    // Arc 共享），所以 factory 里 clone 给出去就行
49    let session_manager = std::sync::Arc::new(LocalSessionManager::default());
50    let mcp_svc = StreamableHttpService::new(
51        {
52            let server = server.clone();
53            move || Ok::<_, std::io::Error>(server.clone())
54        },
55        session_manager,
56        Default::default(),
57    );
58
59    // axum 0.8 router：
60    // - /mcp        → MCP tower service（带 WWW-Authenticate 401 middleware）
61    // - /metrics    → Prometheus 文本
62    // - /.well-known/oauth-protected-resource → RFC 9728 元数据
63    use axum::routing::get;
64    let mcp_with_auth_hint = axum::Router::new()
65        .nest_service("/mcp", mcp_svc)
66        .layer(axum::middleware::from_fn(inject_www_authenticate));
67
68    let app = axum::Router::new()
69        .route(
70            "/metrics",
71            get(|| async {
72                let body = render_mcp_metrics_body();
73                (
74                    axum::http::StatusCode::OK,
75                    [(
76                        axum::http::header::CONTENT_TYPE,
77                        "text/plain; version=0.0.4",
78                    )],
79                    body,
80                )
81            }),
82        )
83        .route(
84            "/.well-known/oauth-protected-resource",
85            get(oauth_protected_resource_metadata),
86        )
87        .merge(mcp_with_auth_hint);
88
89    let bind_addr_sock: std::net::SocketAddr = bind_addr
90        .parse()
91        .map_err(|e| anyhow::anyhow!("invalid bind address {bind_addr}: {e}"))?;
92
93    if let Some((cert_path, key_path)) = tls {
94        // ---------- HTTPS（graceful shutdown 通过 Handle）----------
95        let tls_config =
96            axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert_path, &key_path)
97                .await
98                .with_context(|| {
99                    format!(
100                        "load TLS cert={} key={}",
101                        cert_path.display(),
102                        key_path.display()
103                    )
104                })?;
105        let handle = axum_server::Handle::new();
106        let shutdown_handle = handle.clone();
107        tokio::spawn(async move {
108            shutdown_signal().await;
109            tracing::info!("graceful shutdown: draining HTTPS connections...");
110            shutdown_handle.graceful_shutdown(Some(std::time::Duration::from_secs(10)));
111        });
112        tracing::info!(
113            addr = %bind_addr,
114            cert = %cert_path.display(),
115            "futu-mcp HTTPS transport started \
116             (MCP: /mcp, metrics: /metrics, OAuth metadata: /.well-known/oauth-protected-resource)"
117        );
118        axum_server::bind_rustls(bind_addr_sock, tls_config)
119            .handle(handle)
120            .serve(app.into_make_service())
121            .await
122            .map_err(|e| anyhow::anyhow!("axum-server TLS serve error: {e}"))?;
123    } else {
124        // ---------- plain HTTP（graceful shutdown 通过 axum::serve）----------
125        let listener = tokio::net::TcpListener::bind(&bind_addr)
126            .await
127            .map_err(|e| anyhow::anyhow!("bind {bind_addr}: {e}"))?;
128        tracing::info!(
129            addr = %bind_addr,
130            "futu-mcp HTTP transport started \
131             (MCP: /mcp, metrics: /metrics, OAuth metadata: /.well-known/oauth-protected-resource)"
132        );
133        axum::serve(listener, app)
134            .with_graceful_shutdown(async {
135                shutdown_signal().await;
136                tracing::info!("graceful shutdown: draining HTTP connections...");
137            })
138            .await
139            .map_err(|e| anyhow::anyhow!("axum serve error: {e}"))?;
140    }
141    tracing::info!("server stopped");
142    Ok(())
143}
144
145/// 监听 SIGTERM / SIGINT，任一到达即返回。
146/// 同时兼容 Windows（只有 ctrl_c）和 Unix（SIGTERM + SIGINT）。
147async fn shutdown_signal() {
148    #[cfg(unix)]
149    {
150        use tokio::signal::unix::{SignalKind, signal};
151        let sigterm = match signal(SignalKind::terminate()) {
152            Ok(signal) => Some(signal),
153            Err(e) => {
154                tracing::error!(error = %e, "failed to install SIGTERM handler");
155                None
156            }
157        };
158        let sigint = match signal(SignalKind::interrupt()) {
159            Ok(signal) => Some(signal),
160            Err(e) => {
161                tracing::error!(error = %e, "failed to install SIGINT handler");
162                None
163            }
164        };
165
166        match (sigterm, sigint) {
167            (Some(mut sigterm), Some(mut sigint)) => {
168                tokio::select! {
169                    _ = sigterm.recv() => tracing::info!("received SIGTERM"),
170                    _ = sigint.recv()  => tracing::info!("received SIGINT"),
171                }
172            }
173            (Some(mut sigterm), None) => {
174                sigterm.recv().await;
175                tracing::info!("received SIGTERM");
176            }
177            (None, Some(mut sigint)) => {
178                sigint.recv().await;
179                tracing::info!("received SIGINT");
180            }
181            (None, None) => wait_for_ctrl_c_or_pending().await,
182        }
183    }
184    #[cfg(not(unix))]
185    {
186        wait_for_ctrl_c_or_pending().await;
187    }
188}
189
190async fn wait_for_ctrl_c_or_pending() {
191    match tokio::signal::ctrl_c().await {
192        Ok(()) => tracing::info!("received Ctrl-C"),
193        Err(e) => {
194            tracing::error!(
195                error = %e,
196                "failed to install ctrl-c handler; graceful shutdown signal unavailable"
197            );
198            std::future::pending::<()>().await;
199        }
200    }
201}
202
203/// RFC 9728 — OAuth 2.0 Protected Resource Metadata
204///
205/// 我们不是完整 OAuth 授权服务器（那需要独立的 IdP），只是告诉 MCP 客户端
206/// "这个资源要 Bearer token，scope 列表如下"。LLM agent 实际拿到 key 的方式
207/// 仍然是运维线下发放 + 写入 MCP client 配置里的 `Authorization` 头。
208///
209/// 客户端可以 GET `/.well-known/oauth-protected-resource` 来发现：
210///   - `resource`:              本 MCP endpoint URI
211///   - `bearer_methods_supported`: 我们只支持 `header`（Authorization: Bearer ...）
212///   - `scopes_supported`:      可声明的 futu-auth scope 列表
213///   - `resource_name` / `resource_documentation`: 给人看的说明
214pub(super) async fn oauth_protected_resource_metadata() -> axum::response::Json<serde_json::Value> {
215    axum::response::Json(serde_json::json!({
216        "resource": "/mcp",
217        "bearer_methods_supported": ["header"],
218        "scopes_supported": [
219            "qot:read",
220            "acc:read",
221            "trade:simulate",
222            "trade:real",
223            "trade:unlock"
224        ],
225        "resource_name": "FutuOpenD-rs MCP",
226        "resource_documentation": "https://futuapi.com/reference/mcp/",
227    }))
228}
229
230/// Tower middleware: 如果下游（MCP service）返回 401/403 又没 `WWW-Authenticate`
231/// 头，补一个 `Bearer resource_metadata="..."`，指向 `/.well-known/oauth-protected-resource`。
232///
233/// 符合 RFC 9728 §5.1：资源服务器应通过 WWW-Authenticate 宣告 metadata URL，
234/// 让未配置的客户端能自动发现 scope 和鉴权方式。
235pub(super) async fn inject_www_authenticate(
236    req: axum::extract::Request,
237    next: axum::middleware::Next,
238) -> axum::response::Response {
239    let mut resp = next.run(req).await;
240    let status = resp.status();
241    if (status == axum::http::StatusCode::UNAUTHORIZED
242        || status == axum::http::StatusCode::FORBIDDEN)
243        && !resp
244            .headers()
245            .contains_key(axum::http::header::WWW_AUTHENTICATE)
246    {
247        // 相对路径 —— 客户端按 Host header 拼全 URL；也避免 TLS 终止在前置
248        // 反代（Caddy / Nginx）时我们误把内网地址写进响应头
249        let value = axum::http::HeaderValue::from_static(
250            "Bearer resource_metadata=\"/.well-known/oauth-protected-resource\"",
251        );
252        resp.headers_mut()
253            .insert(axum::http::header::WWW_AUTHENTICATE, value);
254    }
255    resp
256}
futu_mcp/http.rs

futu_mcp/
http.rs
