futu_mcp/

main.rs

//! FutuOpenD-rs MCP 服务器
//!
//! 通过 Model Context Protocol 把 Futu 行情/账户能力暴露给 Claude / LLM 客户端。
//!
//! 授权有两种模式：
//!
//! - **Scope 模式**：`--keys-file <path>` 启用。stdio 客户端通过
//!   `FUTU_MCP_API_KEY` 提供进程级 key；HTTP 客户端的每个 `/mcp` 请求必须
//!   携带有效 `Authorization: Bearer`。服务器按 scope + 限额放行。
//! - **Legacy 模式**：未提供 keys-file 时回退到旧的
//!   `--enable-trading` / `--allow-real-trading` 两级开关。

mod card_num_expand;
mod guard;
mod handlers;
mod http;
mod qot_sdk_adapter;
mod state;
mod tool_account;
mod tool_args;
mod tool_auth;
mod tool_enums;
mod tools;
mod trade_pwd;
mod trd_sdk_adapter;
// v1.4.90 P0-A: resilient stdio transport — recovers from malformed JSON
// (e.g. `{"price": Infinity}`) instead of `exit(0)`-ing the whole server.
// See crates/futu-mcp/src/transport.rs for full rationale.
mod transport;

use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context, Result};
use clap::{ArgMatches, CommandFactory, FromArgMatches, Parser, parser::ValueSource};
use futu_auth::KeyStore;
use rmcp::ServiceExt;
// v1.4.90 P0-A: stdio() (rmcp default) treats parse errors as fatal — see transport.rs.
use crate::transport::resilient_stdio;
use tracing_subscriber::{
    EnvFilter, Layer, filter::filter_fn, fmt, layer::SubscriberExt, util::SubscriberInitExt,
};

#[cfg(test)]
pub(crate) use crate::card_num_expand::build_card_num_resolver;
use crate::card_num_expand::spawn_card_num_expand_retry;
#[cfg(unix)]
use crate::card_num_expand::spawn_sighup_reload;
use crate::http::serve_http;
#[cfg(test)]
use crate::http::{
    authenticate_mcp_transport, inject_www_authenticate, oauth_protected_resource_metadata,
    render_mcp_metrics_body_for,
};
use crate::state::ServerState;
use crate::tools::FutuServer;

#[derive(Default)]
struct RmcpDrainEventFields {
    message: Option<String>,
    error: Option<String>,
}

impl tracing::field::Visit for RmcpDrainEventFields {
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        match field.name() {
            "message" => self.message = Some(value.to_string()),
            "error" => self.error = Some(value.to_string()),
            _ => {}
        }
    }

    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        let rendered = format!("{value:?}");
        let rendered = rendered
            .strip_prefix('"')
            .and_then(|value| value.strip_suffix('"'))
            .unwrap_or(&rendered)
            .to_string();
        match field.name() {
            "message" => self.message = Some(rendered),
            "error" => self.error = Some(rendered),
            _ => {}
        }
    }
}

fn is_expected_rmcp_drain_close(
    target: &str,
    level: &tracing::Level,
    message: Option<&str>,
    error: Option<&str>,
) -> bool {
    target == "rmcp::service"
        && *level == tracing::Level::ERROR
        && message == Some("failed to send pending response during drain")
        && error == Some("channel closed")
}

#[derive(Clone, Copy)]
struct SuppressExpectedRmcpDrainClose;

impl<S> tracing_subscriber::layer::Filter<S> for SuppressExpectedRmcpDrainClose
where
    S: tracing::Subscriber,
{
    fn enabled(
        &self,
        _meta: &tracing::Metadata<'_>,
        _cx: &tracing_subscriber::layer::Context<'_, S>,
    ) -> bool {
        true
    }

    fn event_enabled(
        &self,
        event: &tracing::Event<'_>,
        _cx: &tracing_subscriber::layer::Context<'_, S>,
    ) -> bool {
        let meta = event.metadata();
        if meta.target() != "rmcp::service" || *meta.level() != tracing::Level::ERROR {
            return true;
        }
        let mut fields = RmcpDrainEventFields::default();
        event.record(&mut fields);
        !is_expected_rmcp_drain_close(
            meta.target(),
            meta.level(),
            fields.message.as_deref(),
            fields.error.as_deref(),
        )
    }
}

/// 初始化 stderr 日志 + 可选 audit JSONL 层
///
/// - 常规事件走 stderr（no-ansi，因为 MCP client 的 stderr 往往不是 tty）
/// - 如果 `audit_path` 传了，加一个 target=futu_audit 的 JSON 层写到文件/目录
fn setup_logging(
    default_level: &str,
    audit_path: Option<&std::path::Path>,
) -> Result<Option<tracing_appender::non_blocking::WorkerGuard>> {
    let filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default_level));

    let fmt_layer = fmt::layer()
        .with_timer(futu_core::log::LocalRfc3339Timer)
        .with_writer(std::io::stderr)
        .with_ansi(false)
        .with_filter(SuppressExpectedRmcpDrainClose);

    let registry = tracing_subscriber::registry().with(filter).with(fmt_layer);

    if let Some(path) = audit_path {
        let (writer, guard) = futu_auth::audit::open_writer(path)
            .with_context(|| format!("open audit log {}", path.display()))?;
        let audit_layer = fmt::layer()
            .json()
            .with_timer(futu_core::log::LocalRfc3339Timer)
            .flatten_event(true)
            .with_current_span(false)
            .with_span_list(false)
            .with_target(true)
            .with_writer(writer)
            .with_filter(filter_fn(|meta| meta.target() == futu_auth::audit::TARGET));
        registry.with(audit_layer).init();
        tracing::info!(
            path = %path.display(),
            "audit JSONL logger enabled (target=futu_audit)"
        );
        Ok(Some(guard))
    } else {
        registry.init();
        Ok(None)
    }
}

/// FutuOpenD-rs MCP server
#[derive(Parser)]
#[command(
    name = "futu-mcp",
    version,
    about = "FutuOpenD-rs MCP server",
    long_about = "通过 Model Context Protocol 暴露 Futu 行情/账户工具。默认 stdio transport。"
)]
struct Cli {
    /// 网关地址（可用 FUTU_GATEWAY 环境变量覆盖）
    #[arg(short, long, env = "FUTU_GATEWAY", default_value = "127.0.0.1:11111")]
    gateway: String,

    /// 启用 debug 日志
    #[arg(short, long)]
    verbose: bool,

    /// Scope 模式：加载 keys.json 文件（API Key 授权）。
    ///
    /// stdio 模式需通过 FUTU_MCP_API_KEY / --api-key 提供默认身份；HTTP 模式
    /// 则要求每个 /mcp 请求携带有效 Authorization Bearer。scope / 限额由
    /// keys.json 配置决定；此时 --enable-trading / --allow-real-trading 被忽略。
    #[arg(long)]
    keys_file: Option<PathBuf>,

    /// stdio 进程级 API Key 明文（等价于 FUTU_MCP_API_KEY 环境变量）。
    /// HTTP scope 模式不会用它替代请求自己的 Authorization Bearer。
    ///
    /// 生产环境强烈建议用环境变量而非命令行参数（后者会进 `ps` 输出）。
    #[arg(long, env = "FUTU_MCP_API_KEY", hide_env_values = true)]
    api_key: Option<String>,

    /// Authenticated OpenD REST base URL used by guarded real-money group
    /// writes when daemon raw TCP is disabled by secure KeyStore mode.
    #[arg(long, env = "FUTU_MCP_OPEND_REST_URL")]
    opend_rest_url: Option<String>,

    /// 交易密码所属登录账号，用于兼容读取账号级 credential-store 条目。
    ///
    /// 读取优先级为 FUTU_TRADE_PWD（trim 后非空）> 账号级条目 > 旧全局条目；
    /// 未设置账号时会尝试 FUTU_ACCOUNT 作为账号 hint。macOS 发布包不支持
    /// `futucli` 跨 binary 写入，MCP 请使用 FUTU_TRADE_PWD。
    #[arg(long, env = "FUTU_TRADE_PWD_ACCOUNT")]
    trade_pwd_account: Option<String>,

    /// [Legacy] 启用交易写工具（place / modify / cancel）。默认关闭。
    ///
    /// 开启后默认仅允许 simulate 环境；要操作真实账户需额外 --allow-real-trading。
    /// 注意：下单前网关必须已 unlock_trade（密码不经过 MCP / LLM）。
    /// 若提供了 --keys-file，此开关被忽略，改由 key 的 scope 决定。
    #[arg(long)]
    enable_trading: bool,

    /// [Legacy] 允许交易写工具对 real 环境执行。必须与 --enable-trading 搭配。
    #[arg(long, requires = "enable_trading")]
    allow_real_trading: bool,

    /// 审计日志输出：JSONL 文件路径或目录
    ///
    /// - 带扩展名（`/var/log/futu-mcp-audit.jsonl`）→ 单文件 append
    /// - 不带扩展名 / 以 `/` 结尾 → 每日滚动 `futu-audit.log` + 日期
    ///
    /// 只记录 auth / 交易 事件（target = `futu_audit`）。
    #[arg(long)]
    audit_log: Option<PathBuf>,

    /// 以 HTTP transport 启动（streamable HTTP），监听该端口（格式 `host:port` 或 `:port`）
    ///
    /// 默认 stdio：LLM 客户端启子进程走 stdin/stdout。开 HTTP 后可以让多个
    /// 客户端连同一个 MCP 进程，并同时暴露 `/metrics`。HTTP 调用统一通过
    /// `Authorization: Bearer <api-key>` 鉴权；只有 schema 显式声明 `api_key`
    /// 的工具才支持参数级 key 覆盖。
    ///
    /// 例：`--http-listen 127.0.0.1:3000` / `--http-listen :3000`
    #[arg(long)]
    http_listen: Option<String>,

    /// TLS 证书文件路径（PEM 格式；需与 --tls-key 配合）
    ///
    /// 启用后 HTTP transport 走 HTTPS。若不设置，走纯 HTTP（建议前置 Caddy / Nginx
    /// 做 TLS 终止）。
    #[arg(long, requires = "tls_key")]
    tls_cert: Option<PathBuf>,

    /// TLS 私钥文件路径（PEM 格式；需与 --tls-cert 配合）
    #[arg(long, requires = "tls_cert")]
    tls_key: Option<PathBuf>,

    /// TOML 配置文件路径（字段名与 CLI 参数一致，CLI 参数覆盖配置文件）
    ///
    /// 示例：
    /// ```toml
    /// gateway = "10.0.0.1:11111"
    /// http_listen = ":3000"
    /// keys_file = "/etc/futu/keys.json"
    /// audit_log = "/var/log/futu-mcp-audit.jsonl"
    /// tls_cert = "/etc/futu/cert.pem"
    /// tls_key  = "/etc/futu/key.pem"
    /// ```
    #[arg(long)]
    config: Option<PathBuf>,
}

/// TOML 配置文件映射——字段名与 CLI 参数完全一致
///
/// codex 0547 F4 (P2) fix: 加 `#[serde(deny_unknown_fields)]` — 与 `futu-opend`
/// XmlConfig (BUG-006 v1.4.102 加的) 同语义级别. typo (e.g. `keys_flie` /
/// `auditlog`) 之前 silent drop, 用户配置 silent 失效:
/// - `key_file` typo → keystore 不加载 → MCP 进 legacy mode (无 scope)
/// - `http_litsen` typo → HTTP transport 不启动 (默认 stdio)
/// - `auditlog` typo → 审计文件无事件
///
/// 修后: 任何 unknown field / typo / `[server]` 类未支持 section 立即 parse
/// fatal, daemon abort + 清晰错误.
///
/// **不 break 老 deprecated alias**: 没有 alias 历史, MCP TOML schema 自 v1.0
/// 起字段名稳定, 升级用户无 typo 不会被影响.
#[derive(Debug, Default, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
struct FileConfig {
    gateway: Option<String>,
    verbose: Option<bool>,
    keys_file: Option<PathBuf>,
    api_key: Option<String>,
    opend_rest_url: Option<String>,
    trade_pwd_account: Option<String>,
    enable_trading: Option<bool>,
    allow_real_trading: Option<bool>,
    audit_log: Option<PathBuf>,
    http_listen: Option<String>,
    tls_cert: Option<PathBuf>,
    tls_key: Option<PathBuf>,
}

/// codex 0547 F5 (P3) fix: clap `ValueSource` 区分 "CLI 显式传" vs "默认值".
///
/// 之前用 `self.gateway == "127.0.0.1:11111"` 判 "CLI 没传" — 当用户**显式**
/// 传 `--gateway 127.0.0.1:11111` (与默认值相等) 时被错判为 "未传" → TOML
/// 配置 gateway override 反向. 违背 "CLI 始终覆盖配置文件" 契约.
///
/// 同模式: bool 字段 (`verbose` / `enable_trading` / `allow_real_trading`)
/// 之前用 `if !self.field` 判, 用户显式 `--verbose` 时反复 = false 也不能
/// 区分 "CLI 没传 + TOML 也没设" 与 "CLI 显式 false (无 --no-flag)". clap
/// derive 不天然支持 `--no-*`, 所以 bool 字段的 explicit-false override 是
/// 设计 limitation; 但 explicit-true 一定要尊重.
///
/// 本 helper 接受 `&ArgMatches` 与 `arg_id`, 返 true 仅在 user explicitly 传
/// (而不是 default / env). 见
/// <https://docs.rs/clap/latest/clap/parser/enum.ValueSource.html>.
fn is_cli_explicit(matches: &ArgMatches, arg_id: &str) -> bool {
    matches!(
        matches.value_source(arg_id),
        Some(ValueSource::CommandLine) | Some(ValueSource::EnvVariable)
    )
}

impl Cli {
    /// 如果指定了 `--config`，先从文件读取默认值，再让 CLI 参数覆盖。
    ///
    /// codex 0547 F5 (P3): 用 `&ArgMatches` 精准判断 "CLI/env 是否显式传"
    /// 而非旧的 "值 == 默认 → 当作没传" 启发式. CLI 显式传 = 显式 (即使值
    /// 等于默认). TOML 文件值仅在 CLI / env 都没显式传时才采用.
    fn merge_config(mut self, matches: &ArgMatches) -> Result<Self> {
        let Some(config_path) = &self.config else {
            return Ok(self);
        };
        let content = std::fs::read_to_string(config_path)
            .with_context(|| format!("read config file {}", config_path.display()))?;
        let fc: FileConfig = toml::from_str(&content)
            .with_context(|| format!("parse config file {}", config_path.display()))?;

        // codex 0547 F5: gateway 用 ValueSource 精准判. 之前 "self.gateway ==
        // 默认值" 启发式在用户**显式**传默认值时反向 (TOML 覆盖 CLI).
        if let Some(g) = fc.gateway
            && !is_cli_explicit(matches, "gateway")
        {
            self.gateway = g;
        }
        // codex 0547 F5: Option 字段用 None check (CLI 没传 = None, 不会与
        // 默认值 ambiguity).
        if self.keys_file.is_none() {
            self.keys_file = fc.keys_file;
        }
        if self.api_key.is_none()
            && let Some(k) = fc.api_key
        {
            self.api_key = Some(k);
        }
        if self.opend_rest_url.is_none() {
            self.opend_rest_url = fc.opend_rest_url;
        }
        if self.trade_pwd_account.is_none() {
            self.trade_pwd_account = fc.trade_pwd_account;
        }
        // codex 0547 F5: bool 字段用 ValueSource 区分 "未传" vs "显式 false".
        // clap derive 没 `--no-verbose`, 所以无法 explicit-set false; 但用户
        // **显式 true** (e.g. `--verbose` 在 CLI) 要保留 (TOML 即使写 false
        // 也不能覆盖 explicit-true).
        if fc.verbose.is_some() && !is_cli_explicit(matches, "verbose") {
            self.verbose = fc.verbose.unwrap_or(false);
        }
        if fc.enable_trading.is_some() && !is_cli_explicit(matches, "enable_trading") {
            self.enable_trading = fc.enable_trading.unwrap_or(false);
        }
        if fc.allow_real_trading.is_some() && !is_cli_explicit(matches, "allow_real_trading") {
            self.allow_real_trading = fc.allow_real_trading.unwrap_or(false);
        }
        if self.audit_log.is_none() {
            self.audit_log = fc.audit_log;
        }
        if self.http_listen.is_none() {
            self.http_listen = fc.http_listen;
        }
        if self.tls_cert.is_none() {
            self.tls_cert = fc.tls_cert;
        }
        if self.tls_key.is_none() {
            self.tls_key = fc.tls_key;
        }
        // 此时 tracing 可能还没初始化，写 stderr 即可
        eprintln!("[config] loaded {}", config_path.display());
        Ok(self)
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // codex 0547 F5 (P3): 解析两次 — 用 ArgMatches 区分 explicit vs default
    // 后再用 derive 反向 build Cli 结构. 单次 parse 走不通 (Cli::parse 不暴露
    // ArgMatches), 但解析+from_arg_matches 是 0-allocation cycle.
    let matches = Cli::command().get_matches();
    let cli = Cli::from_arg_matches(&matches)
        .map_err(|e| anyhow::anyhow!("clap derive build failed: {e}"))?
        .merge_config(&matches)?;

    // MCP 用 stdout 传协议帧，所有日志必须写 stderr
    let default_level = if cli.verbose { "debug" } else { "info" };

    // audit 日志 guard 必须活到 main 返回；否则后台 flush 可能丢事件
    let _audit_guard = setup_logging(default_level, cli.audit_log.as_deref())?;

    // ---------- 加载 KeyStore ----------
    let key_store = match &cli.keys_file {
        Some(path) => {
            let store = KeyStore::load(path)
                .with_context(|| format!("load keys file {}", path.display()))?;
            tracing::info!(
                path = %path.display(),
                keys_loaded = store.len(),
                "scope mode: keys file loaded"
            );
            if cli.enable_trading || cli.allow_real_trading {
                tracing::warn!(
                    "--enable-trading / --allow-real-trading are IGNORED in scope mode; \
                     trading permissions are controlled by API key scopes"
                );
            }
            Arc::new(store)
        }
        None => {
            tracing::info!("legacy mode: no keys file; using --enable-trading switches");
            Arc::new(KeyStore::empty())
        }
    };

    // ---------- 校验调用方 API key ----------
    let http_mode = cli.http_listen.is_some();
    let authed_key = if key_store.is_configured() && http_mode {
        if cli.api_key.as_deref().is_some_and(|key| !key.is_empty()) {
            tracing::warn!(
                "--api-key / FUTU_MCP_API_KEY is stdio-only and IGNORED in HTTP scope mode; \
                 every /mcp request must provide its own Authorization Bearer"
            );
        }
        None
    } else if key_store.is_configured() {
        match cli.api_key.as_deref() {
            Some(plaintext) if !plaintext.is_empty() => match key_store.verify(plaintext) {
                Some(rec) => {
                    tracing::info!(
                        key_id = %rec.id,
                        scopes = ?rec.scopes.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
                        "API key verified"
                    );
                    Some(rec)
                }
                None => {
                    tracing::error!("FUTU_MCP_API_KEY does not match any key in keys.json");
                    None
                }
            },
            _ => {
                tracing::warn!(
                    "stdio scope mode active but FUTU_MCP_API_KEY not set; \
                     all tool calls will be rejected"
                );
                None
            }
        }
    } else {
        None
    };

    tracing::info!(
        gateway = %cli.gateway,
        scope_mode = key_store.is_configured(),
        enable_trading = cli.enable_trading,
        allow_real_trading = cli.allow_real_trading,
        trade_pwd_account = cli.trade_pwd_account.as_deref().unwrap_or("<legacy/env>"),
        "futu-mcp starting"
    );
    if !key_store.is_configured() && cli.enable_trading {
        tracing::warn!(
            allow_real_trading = cli.allow_real_trading,
            "trading write tools ENABLED (legacy mode)"
        );
    }

    let state = ServerState::new(cli.gateway)
        .with_trading(cli.enable_trading, cli.allow_real_trading)
        .with_key_store(key_store.clone())
        .with_authed_key(authed_key)
        .with_opend_rest(cli.opend_rest_url, cli.api_key.clone())
        .with_trade_pwd_account(cli.trade_pwd_account);
    let server = FutuServer::new(state.clone());

    // v1.4.105 external reviewer #4 (BUG-v1.4.104-002, P1) fix: standalone MCP 启动时把
    // `allowed_card_nums` (string format, e.g. ["0757"]) resolve 成
    // `allowed_acc_ids` (numeric set), 行为与 futu-opend daemon 启动 +
    // SIGHUP 路径**byte-identical** (resolver 4-suffix / 16-exact 双匹配
    // card_num + uni_card_num, 1 个 → resolved, 0 → unresolved warn,
    // ≥2 → ambiguous warn). 不做 expand 时 KeyStore::load_file 注入的
    // fail-closed sentinel `allowed_acc_ids = {0}` 会让真账户 acc_id ≠ 0
    // 永远 reject "not in allowed list {0}".
    //
    // 设计要点:
    // - 仅在 KeyStore 至少一条 key 配置了 allowed_card_nums 时才连 daemon
    //   (避免无意义 GetAccList 请求)
    // - daemon 可能尚未起来 / connect race → 后台 task 重试 6 次 × 10s
    //   覆盖 daemon 启动 ~60s 窗口, 与 daemon 内部 trd_cache 加载 retry 同节奏
    // - expand 失败不阻塞 MCP server 启动 — sentinel 仍生效保护
    // - SIGHUP 重载 keys.json 后必须重新 expand (sentinel/旧 acc_ids 会失效)
    if key_store.is_configured() && key_store.has_any_card_num_restrictions() {
        spawn_card_num_expand_retry(state.clone(), key_store.clone());
    } else if key_store.is_configured() {
        tracing::debug!(
            "v1.4.105 external report #4: keystore 无 allowed_card_nums 限制, 跳过 daemon expand"
        );
    }

    // SIGHUP 热重载 keys.json（unix only）— v1.4.105 external reviewer #4: reload 后 +
    // re-expand card_num (与 daemon `card_num_reload_and_expand_fn` 同语义)
    #[cfg(unix)]
    spawn_sighup_reload(key_store.clone(), state.clone());

    // v1.0：install 全局 metrics registry，让 audit::* 的 counter hook 起作用
    // （HTTP 模式下 /metrics 端点消费这套；stdio 模式虽然没 HTTP，但写进内存
    // 方便 debug 和后续加 transport）
    futu_auth::metrics::install(std::sync::Arc::new(futu_auth::MetricsRegistry::default()));

    if let Some(listen) = cli.http_listen {
        let tls = match (cli.tls_cert, cli.tls_key) {
            (Some(cert), Some(key)) => Some((cert, key)),
            _ => None,
        };
        serve_http(server, key_store, &listen, tls).await?;
    } else {
        serve_stdio(server).await?;
    }

    Ok(())
}

/// stdio 模式：MCP 客户端启动子进程，stdin/stdout 传协议帧
///
/// v1.4.90 P0-A: uses `resilient_stdio()` instead of `rmcp::transport::stdio()`
/// — a malformed JSON line (e.g. `{"price": Infinity}` from an LLM client)
/// now produces a `-32700 Parse error` response instead of `exit(0)`-ing the
/// entire server. See crates/futu-mcp/src/transport.rs for full background.
async fn serve_stdio(server: tools::FutuServer) -> Result<()> {
    let transport = resilient_stdio();
    let drain = transport.pending_response_drain();
    let service = match server.serve(transport).await {
        Ok(service) => service,
        Err(error) => {
            // rmcp can return an initialize error before the background stdout
            // writer flushes its JSON-RPC error frame. Keep the runtime alive
            // until the transport confirms every accepted response ID was
            // written, or the existing bounded drain grace expires.
            if !drain.wait_pending_responses().await {
                tracing::warn!(
                    "MCP stdio: initialize response did not drain before shutdown grace expired"
                );
            }
            return Err(anyhow::anyhow!("MCP service init failed: {error}"));
        }
    };

    if let Err(error) = service.waiting().await {
        if !drain.wait_pending_responses().await {
            tracing::warn!(
                "MCP stdio: response did not drain before service shutdown grace expired"
            );
        }
        return Err(anyhow::anyhow!("MCP service error: {error}"));
    }
    Ok(())
}

#[cfg(test)]
mod tests;