futu_mcp/

main.rs

//! FutuOpenD-rs MCP 服务器
//!
//! 通过 Model Context Protocol 把 Futu 行情/账户能力暴露给 Claude / LLM 客户端。
//!
//! 授权有两种模式：
//!
//! - **Scope 模式**：`--keys-file <path>` 启用，客户端必须通过 `FUTU_MCP_API_KEY`
//!   环境变量传入明文 key。服务器用 SHA-256 hash 比对 keys.json 中的记录，
//!   按 scope + 限额放行。
//! - **Legacy 模式**：未提供 keys-file 时回退到旧的
//!   `--enable-trading` / `--allow-real-trading` 两级开关。

mod guard;
mod handlers;
mod state;
mod tools;

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

use anyhow::{Context, Result};
use clap::Parser;
use futu_auth::KeyStore;
use rmcp::{transport::stdio, ServiceExt};
use tracing_subscriber::{
    filter::filter_fn, fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer,
};

use crate::state::ServerState;
use crate::tools::FutuServer;

/// 初始化 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_writer(std::io::stderr).with_ansi(false);

    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()
            .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 授权）。
    ///
    /// 启用后所有工具调用必须带 FUTU_MCP_API_KEY 环境变量，
    /// scope / 限额由 keys.json 配置决定；此时 --enable-trading / --allow-real-trading 被忽略。
    #[arg(long)]
    keys_file: Option<PathBuf>,

    /// 调用方 API Key 明文（等价于 FUTU_MCP_API_KEY 环境变量）
    ///
    /// 生产环境强烈建议用环境变量而非命令行参数（后者会进 `ps` 输出）。
    #[arg(long, env = "FUTU_MCP_API_KEY", hide_env_values = true)]
    api_key: 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`。per-call key 覆盖依然
    /// 走 tool args 的 `api_key` 字段；HTTP-layer 的 Authorization header 未来
    /// 版本再接（v1.0 先做传输层切换）。
    ///
    /// 例：`--http-listen 127.0.0.1:3000` / `--http-listen :3000`
    #[arg(long)]
    http_listen: Option<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();

    // 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 authed_key = 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!(
                    "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,
        "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);
    let server = FutuServer::new(state);

    // SIGHUP 热重载 keys.json（unix only）
    #[cfg(unix)]
    spawn_sighup_reload(key_store);

    // 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 {
        serve_http(server, &listen).await?;
    } else {
        serve_stdio(server).await?;
    }

    Ok(())
}

/// stdio 模式：MCP 客户端启动子进程，stdin/stdout 传协议帧
async fn serve_stdio(server: tools::FutuServer) -> Result<()> {
    let service = server
        .serve(stdio())
        .await
        .map_err(|e| anyhow::anyhow!("MCP service init failed: {e}"))?;

    service
        .waiting()
        .await
        .map_err(|e| anyhow::anyhow!("MCP service error: {e}"))?;
    Ok(())
}

/// HTTP 模式：axum + rmcp StreamableHttpService，`/mcp` 路径跑 MCP，
/// `/metrics` 暴露 Prometheus counters（无需 token）。
async fn serve_http(server: tools::FutuServer, listen: &str) -> Result<()> {
    use rmcp::transport::streamable_http_server::{
        session::local::LocalSessionManager, StreamableHttpService,
    };

    // 补齐 `:port` 写法：bind 到 0.0.0.0:port
    let bind_addr = if listen.starts_with(':') {
        format!("0.0.0.0{listen}")
    } else {
        listen.to_string()
    };

    // rmcp StreamableHttpService 要求一个 service factory —— 每个 HTTP 会话要
    // 一份独立的 MCP Service 实例。FutuServer 目前是 Clone 的（ServerState 内
    // Arc 共享），所以 factory 里 clone 给出去就行
    let session_manager = std::sync::Arc::new(LocalSessionManager::default());
    let mcp_svc = StreamableHttpService::new(
        {
            let server = server.clone();
            move || Ok::<_, std::io::Error>(server.clone())
        },
        session_manager,
        Default::default(),
    );

    // axum 0.8 router：/mcp 挂 MCP tower service，/metrics 独立路由
    use axum::routing::get;
    let app = axum::Router::new()
        .route(
            "/metrics",
            get(|| async {
                let body = futu_auth::metrics::global()
                    .map(|r| r.render_prometheus())
                    .unwrap_or_default();
                (
                    axum::http::StatusCode::OK,
                    [(
                        axum::http::header::CONTENT_TYPE,
                        "text/plain; version=0.0.4",
                    )],
                    body,
                )
            }),
        )
        .nest_service("/mcp", mcp_svc);

    let listener = tokio::net::TcpListener::bind(&bind_addr)
        .await
        .map_err(|e| anyhow::anyhow!("bind {bind_addr}: {e}"))?;
    tracing::info!(
        addr = %bind_addr,
        "futu-mcp HTTP transport started (MCP: /mcp, metrics: /metrics)"
    );
    axum::serve(listener, app)
        .await
        .map_err(|e| anyhow::anyhow!("axum serve error: {e}"))?;
    Ok(())
}

#[cfg(unix)]
fn spawn_sighup_reload(store: Arc<KeyStore>) {
    if !store.is_configured() {
        return;
    }
    use tokio::signal::unix::{signal, SignalKind};
    tokio::spawn(async move {
        let mut sig = match signal(SignalKind::hangup()) {
            Ok(s) => s,
            Err(e) => {
                tracing::error!(error = %e, "failed to install SIGHUP handler");
                return;
            }
        };
        tracing::info!("SIGHUP handler installed; send `kill -HUP <pid>` to reload keys");
        while sig.recv().await.is_some() {
            match store.reload() {
                Ok(()) => tracing::warn!(keys_loaded = store.len(), "keys file reloaded on SIGHUP"),
                Err(e) => tracing::error!(error = %e, "SIGHUP reload failed; keeping old keys"),
            }
        }
    });
}