alkhttp 0.5.0

//! `from_mcp`: discover remote MCP tools over streamable HTTP and register
//! each as a [`HandlerRegistration`] bundle with a forwarding handler that
//! calls the remote tool via `tools/call` (ADR-041 for the inverse
//! projection; this is the import direction).
//!
//! Streamable HTTP only (ADR-037 — stdio is not built). Feature-gated behind
//! `mcp`. **The credential is captured at import time:** the token given to
//! [`FromMCP::with_auth_token`] pins the rmcp transport's auth header for the
//! session's lifetime; per-call credentials handed to
//! `OperationContext.capabilities` are not observed by the forwarding
//! handler (the no-env-vars invariant, ADR-014, still holds — the token
//! comes from config, never `std::env::var`). Provenance is `FromMCP`
//! (leaf — `composition_authority: None`, `scoped_env: None`, `Internal`
//! by default — ADR-015/022).
//!
//! Session teardown (explicit limitation, review-001 CON-08): `import()`
//! detaches the rmcp client session fire-and-forget; there is no close
//! handle on [`FromMCP`], so each import leaves the remote server-side
//! session and its GET SSE stream open until the remote times it out.
//! Repeated imports accumulate sessions — reconnecting importers should
//! import once per process (or wait for a teardown handle in a later
//! version).
//!
//! Tool discovery is a bounded pagination loop over `tools/list` — never
//! rmcp's unbounded `Peer::list_all_tools`: at most `MCP_MAX_TOOLS_LIST_PAGES`
//! pages within `MCP_TOOLS_LIST_DEADLINE` overall. Trip either budget and
//! the import fails loudly with `AdapterError::DiscoveryFailed` naming the
//! budget (pages fetched, tools accumulated) — no silent truncation, no
//! partial registration (review-002 CON-14).

use std::time::{Duration, Instant};

use alkcall::client::{AdapterError, OperationAdapter};
use alkcall::core::types::{Capabilities, Secret};
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::context::OperationContext;
use alkcall::registry::registration::{
    make_handler, HandlerKind, HandlerRegistration, OperationProvenance,
};
use alkcall::registry::spec::{
    AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility,
};
use rmcp::model::{
    CallToolRequestParams, CallToolResult, ClientCapabilities, ClientInfo, Content, Implementation,
    JsonObject, PaginatedRequestParams, Tool,
};
use rmcp::service::RoleClient;
use rmcp::transport::{
    streamable_http_client::{StreamableHttpClientTransportConfig, StreamableHttpError},
    DynamicTransportError, StreamableHttpClientTransport,
};
use rmcp::{Peer, ServiceError, ServiceExt};
use serde_json::{Map, Value};

const MCP_CAPABILITY_KEY: &str = "mcp";

/// Upper bound on `tools/list` pages a single `import()` will fetch.
const MCP_MAX_TOOLS_LIST_PAGES: u32 = 100;

/// Overall time budget for the full `tools/list` pagination walk.
const MCP_TOOLS_LIST_DEADLINE: Duration = Duration::from_secs(60);

/// Declared error for transport-level `tools/call` failures (review-001
/// CON-11): the remote is unreachable, the transport closed, or the call
/// timed out. Payload is the array-of-content-blocks shape; retryable.
const MCP_TRANSPORT_ERROR: &str = "MCP_TRANSPORT_ERROR";

/// The `from_mcp` adapter (mcp feature): imports a remote MCP server's
/// tools as call-protocol operations under one namespace, with the
/// auth token injected per-call from `Capabilities` (ADR-014).
pub struct FromMCP {
    endpoint: String,
    auth_token: Option<Secret<String>>,
    namespace: String,
}

impl FromMCP {
    /// Assemble the adapter for a streamable-HTTP MCP `endpoint`,
    /// registering its tools under `namespace`.
    pub fn new(endpoint: impl Into<String>, namespace: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            auth_token: None,
            namespace: namespace.into(),
        }
    }

    /// Set the auth token sent as the MCP endpoint's `Authorization:
    /// Bearer` header (wrapped in `Secret` so it never logs).
    pub fn with_auth_token(mut self, token: impl Into<String>) -> Self {
        self.auth_token = Some(Secret::new(token.into()));
        self
    }

    /// The configured MCP endpoint URL.
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    /// The namespace imported operations register under.
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// The configured token, for introspection (never logged).
    pub fn auth_token(&self) -> Option<&Secret<String>> {
        self.auth_token.as_ref()
    }
}

#[async_trait::async_trait]
impl OperationAdapter for FromMCP {
    async fn import(&self) -> Result<Vec<HandlerRegistration>, AdapterError> {
        let mut config = StreamableHttpClientTransportConfig::with_uri(self.endpoint.clone());
        if let Some(token) = &self.auth_token {
            config = config.auth_header(token.expose_secret().clone());
        }
        let transport = StreamableHttpClientTransport::from_config(config);
        let client_info = ClientInfo::new(
            ClientCapabilities::default(),
            Implementation::new("alkhttp-from-mcp", env!("CARGO_PKG_VERSION")),
        );
        let running = client_info
            .serve(transport)
            .await
            .map_err(|e| classify_init_error(&e))?;
        let peer: Peer<RoleClient> = running.peer().clone();

        let tools = list_all_tools_bounded(&peer).await?;

        let bundles = tools
            .into_iter()
            .map(|tool| build_registration(&peer, &self.namespace, self.auth_token.clone(), tool))
            .collect::<Result<Vec<_>, _>>()?;

        std::mem::forget(running);
        Ok(bundles)
    }
}

/// Bounded `tools/list` pagination walk (review-002 CON-14): the rmcp
/// equivalent loops `while cursor.is_some()` with no page cap and no
/// deadline, so a hostile/buggy server that never clears `next_cursor`
/// hangs `import()` while memory grows without bound. Stops after
/// [`MCP_MAX_TOOLS_LIST_PAGES`] pages or [`MCP_TOOLS_LIST_DEADLINE`]
/// elapses and fails loudly — naming pages fetched and tools accumulated —
/// rather than silently truncating.
async fn list_all_tools_bounded(peer: &Peer<RoleClient>) -> Result<Vec<Tool>, AdapterError> {
    let started = Instant::now();
    let mut tools = Vec::new();
    let mut cursor = None;
    for pages_fetched in 1..=MCP_MAX_TOOLS_LIST_PAGES {
        let remaining = MCP_TOOLS_LIST_DEADLINE
            .checked_sub(started.elapsed())
            .ok_or_else(|| pagination_budget_error(pages_fetched, tools.len()))?;
        let page = match tokio::time::timeout(
            remaining,
            peer.list_tools(Some(PaginatedRequestParams::default().with_cursor(cursor))),
        )
        .await
        {
            Ok(result) => result.map_err(|e| AdapterError::DiscoveryFailed {
                message: format!("tools/list failed: {e}"),
            })?,
            Err(_) => return Err(pagination_budget_error(pages_fetched, tools.len())),
        };
        tools.extend(page.tools);
        cursor = page.next_cursor;
        if cursor.is_none() {
            return Ok(tools);
        }
    }
    Err(pagination_budget_error(
        MCP_MAX_TOOLS_LIST_PAGES,
        tools.len(),
    ))
}

fn pagination_budget_error(pages_fetched: u32, tools_accumulated: usize) -> AdapterError {
    AdapterError::DiscoveryFailed {
        message: format!(
            "tools/list pagination exceeded budget (max {MCP_MAX_TOOLS_LIST_PAGES} pages or \
             {MCP_TOOLS_LIST_DEADLINE:?} overall) after {pages_fetched} page(s) with \
             {tools_accumulated} tool(s) accumulated; import failed without partial registration"
        ),
    }
}

fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError {
    use rmcp::service::ClientInitializeError as E;
    match e {
        E::TransportError { error, .. } => {
            let message = error.to_string();
            if is_unauthorized_transport(error) || auth_error_message(&message) {
                AdapterError::Unauthorized { message }
            } else {
                AdapterError::DiscoveryFailed { message }
            }
        }
        other => AdapterError::DiscoveryFailed {
            message: format!("initialize failed: {other}"),
        },
    }
}

fn is_unauthorized_transport(error: &DynamicTransportError) -> bool {
    match error
        .error
        .downcast_ref::<StreamableHttpError<reqwest::Error>>()
    {
        Some(StreamableHttpError::AuthRequired(_))
        | Some(StreamableHttpError::InsufficientScope(_)) => true,
        Some(StreamableHttpError::Client(e)) => {
            e.status() == Some(reqwest::StatusCode::UNAUTHORIZED)
        }
        _ => false,
    }
}

fn auth_error_message(message: &str) -> bool {
    message.contains("AuthRequired")
        || message.contains("InsufficientScope")
        || message.contains("www-authenticate")
        || message.to_ascii_lowercase().contains("unauthorized")
}

fn build_registration(
    peer: &Peer<RoleClient>,
    namespace: &str,
    auth_token: Option<Secret<String>>,
    tool: Tool,
) -> Result<HandlerRegistration, AdapterError> {
    let spec = build_spec(&tool, namespace)?;
    let caps = capabilities_for(auth_token);

    let tool_name = tool.name.to_string();
    let peer_clone = peer.clone();
    let handler = make_handler(move |input: Value, context: OperationContext| {
        let peer = peer_clone.clone();
        let tool_name = tool_name.clone();
        async move {
            let request_id = context.request_id.clone();
            let arguments = value_to_json_object(input);
            let params = CallToolRequestParams::new(tool_name.clone()).with_arguments(arguments);
            let result = match peer.call_tool(params).await {
                Ok(r) => r,
                Err(e) => {
                    return ResponseEnvelope::error(request_id, transport_call_tool_error(&e))
                }
            };

            map_call_tool_result(result, request_id)
        }
    });

    Ok(HandlerRegistration::new(
        spec,
        HandlerKind::Once(handler),
        OperationProvenance::FromMCP,
        None,
        None,
        caps,
    ))
}

/// Transport-level `tools/call` failure (remote unreachable, timeout,
/// transport closed), mapped to the declared [`MCP_TRANSPORT_ERROR`]
/// schema — never the undeclared `INTERNAL` (review-001 CON-11).
fn transport_call_tool_error(error: &ServiceError) -> CallError {
    let message = format!("tools/call failed: {error}");
    match error {
        ServiceError::McpError(e) => {
            let code = format!("MCP_JRPC_{:+06}", e.code.0);
            let mut err = CallError::new(code, e.message.to_string(), true);
            if let Some(data) = &e.data {
                err = err.with_details(data.clone());
            }
            err
        }
        ServiceError::Timeout { .. }
        | ServiceError::TransportClosed
        | ServiceError::Cancelled { .. }
        | ServiceError::TransportSend(_)
        | ServiceError::UnexpectedResponse => CallError::new(MCP_TRANSPORT_ERROR, message, true),
        _ => CallError::new(MCP_TRANSPORT_ERROR, message, true),
    }
}

fn sanitize_tool_name(tool_name: &str) -> Result<String, AdapterError> {
    let name = tool_name.trim();
    if name.is_empty() {
        return Err(AdapterError::SchemaParse {
            message: "MCP tool name is empty".to_string(),
        });
    }
    if name.contains('/') {
        return Err(AdapterError::SchemaParse {
            message: format!(
                "MCP tool name `{name}` contains `/` — the two-segment ns/op op-name convention \
                 (review-001 CON-12) requires flat tool names; refusing import"
            ),
        });
    }
    if name.chars().any(|c| c.is_whitespace()) {
        return Err(AdapterError::SchemaParse {
            message: format!("MCP tool name `{name}` contains whitespace"),
        });
    }
    Ok(name.to_string())
}

pub(crate) fn build_spec(tool: &Tool, namespace: &str) -> Result<OperationSpec, AdapterError> {
    let tool_name = sanitize_tool_name(&tool.name)?;
    let op_name = format!("{namespace}/{tool_name}");
    let input_schema = json_object_to_value(tool.input_schema.as_ref().clone());
    let output_schema = output_schema_for(tool);
    let error_schemas = error_schemas_for(tool);
    Ok(OperationSpec::new(
        op_name,
        OperationType::Mutation,
        Visibility::Internal,
        input_schema,
        output_schema,
        error_schemas,
        AccessControl::default(),
        None,
    ))
}

pub(crate) fn map_call_tool_result(result: CallToolResult, request_id: String) -> ResponseEnvelope {
    if result.is_error == Some(true) {
        let details = content_blocks_to_value(&result.content);
        let message = if result.content.is_empty() {
            "MCP tool returned isError with no content".to_string()
        } else {
            "MCP tool returned isError".to_string()
        };
        let mut err = CallError::new("MCP_TOOL_ERROR", message, false);
        if details != Value::Null {
            err = err.with_details(details);
        }
        return ResponseEnvelope::error(request_id, err);
    }

    if let Some(structured) = result.structured_content {
        return ResponseEnvelope::ok(request_id, structured);
    }

    let mapped = content_blocks_to_value(&result.content);
    ResponseEnvelope::ok(request_id, mapped)
}

pub(crate) fn output_schema_for(tool: &Tool) -> Value {
    if let Some(schema) = &tool.output_schema {
        json_object_to_value(schema.as_ref().clone())
    } else {
        content_block_union_schema()
    }
}

pub(crate) fn content_block_union_schema() -> Value {
    serde_json::json!({
        "type": "array",
        "description": "MCP ContentBlock union (text | image | audio | resource | resource_link)",
        "items": {
            "oneOf": [
                {
                    "type": "object",
                    "properties": {
                        "type": { "type": "string", "enum": ["text"] },
                        "text": { "type": "string" }
                    },
                    "required": ["type", "text"]
                },
                {
                    "type": "object",
                    "properties": {
                        "type": { "type": "string", "enum": ["image"] },
                        "data": { "type": "string" },
                        "mimeType": { "type": "string" }
                    },
                    "required": ["type", "data", "mimeType"]
                },
                {
                    "type": "object",
                    "properties": {
                        "type": { "type": "string", "enum": ["audio"] },
                        "data": { "type": "string" },
                        "mimeType": { "type": "string" }
                    },
                    "required": ["type", "data", "mimeType"]
                },
                {
                    "type": "object",
                    "properties": {
                        "type": { "type": "string", "enum": ["resource"] },
                        "resource": { "type": "object" }
                    },
                    "required": ["type", "resource"]
                },
                {
                    "type": "object",
                    "properties": {
                        "type": { "type": "string", "enum": ["resource_link"] },
                        "uri": { "type": "string" },
                        "name": { "type": "string" }
                    },
                    "required": ["type", "uri", "name"]
                }
            ]
        }
    })
}

pub(crate) fn content_blocks_to_value(blocks: &[Content]) -> Value {
    let mapped: Vec<Value> = blocks
        .iter()
        .map(|block| serde_json::to_value(block).unwrap_or(Value::Null))
        .collect();
    Value::Array(mapped)
}

fn error_schemas_for(tool: &Tool) -> Vec<ErrorDefinition> {
    vec![
        ErrorDefinition {
            code: "MCP_TOOL_ERROR".to_string(),
            description: format!("MCP tool '{}' reported an error (isError)", tool.name),
            schema: serde_json::json!({
                "type": "array",
                "description": "MCP error content blocks",
                "items": content_block_union_schema()
            }),
            http_status: None,
        },
        ErrorDefinition {
            code: MCP_TRANSPORT_ERROR.to_string(),
            description: format!(
                "the transport failed while calling MCP tool '{}' (remote unreachable, \
                 connection closed, or call timed out); retryable",
                tool.name
            ),
            schema: serde_json::json!({
                "type": "null",
                "description": "transport failures carry no payload"
            }),
            http_status: None,
        },
    ]
}

fn capabilities_for(auth_token: Option<Secret<String>>) -> Capabilities {
    match auth_token {
        Some(token) => {
            Capabilities::new().with_http_token(MCP_CAPABILITY_KEY, token.expose_secret().clone())
        }
        None => Capabilities::new(),
    }
}

fn value_to_json_object(value: Value) -> Map<String, Value> {
    match value {
        Value::Object(map) => map,
        other => {
            let mut map = Map::new();
            map.insert("value".to_string(), other);
            map
        }
    }
}

fn json_object_to_value(map: JsonObject) -> Value {
    Value::Object(map)
}

#[cfg(test)]
mod tests;