prism_mcp_rs/server/

mcp_server.rs

//! MCP server implementation
//!
//! Module provides the main MCP server implementation that handles client connections,
//! manages resources, tools, and prompts, and processes JSON-RPC requests according to
//! the Model Context Protocol specification.
//!
//! # Plugin System Feature
//!
//! To use the plugin system, enable the plugin feature in `Cargo.toml`:
//! ```toml
//! [dependencies]
//! prism-mcp-rs = { version = "3", features = ["plugin"] }
//! ```
//!
//! This enables dynamic tool loading and plugin management capabilities.

use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use tracing::Instrument;

use crate::core::{
    completion::{CompletionContext, CompletionHandler},
    error::{McpError, McpResult},
    prompt::{Prompt, PromptHandler},
    resource::{Resource, ResourceHandler},
    tool::{MultiRoundToolCall, MultiRoundToolHandler, Tool, ToolHandler},
    PromptInfo, ResourceInfo, ToolInfo,
};
use crate::protocol::tasks::{
    has_tasks_extension, CancelTaskParams, CreateTaskResult, GetTaskParams, GetTaskResult,
    TaskAcknowledgement, UpdateTaskParams, TASKS_EXTENSION_ID,
};
use crate::protocol::{messages::*, methods, types::*, validation::*, version::*};
use crate::security::{RequestContext, RequestPolicy, RequestTarget};
use crate::server::tasks::{ComposedTaskToolHandler, TaskRegistry, TaskToolHandler};
use crate::transport::traits::ServerTransport;

#[cfg(feature = "plugin")]
use crate::plugin::PluginManager;
#[cfg(feature = "plugin")]
use async_trait::async_trait;

// Import ClosureWrapper from tool module for closure support
use crate::core::tool::ClosureWrapper;

// Plugin tool handler wrapper
#[cfg(feature = "plugin")]
struct PluginToolHandler {
    manager: Arc<RwLock<PluginManager>>,
    tool_name: String,
}

#[cfg(feature = "plugin")]
impl PluginToolHandler {
    fn new(manager: Arc<RwLock<PluginManager>>, tool_name: String) -> Self {
        Self { manager, tool_name }
    }
}

#[cfg(feature = "plugin")]
#[async_trait]
impl ToolHandler for PluginToolHandler {
    async fn call(
        &self,
        arguments: HashMap<String, Value>,
    ) -> McpResult<crate::protocol::types::ToolResult> {
        self.manager
            .read()
            .await
            .execute_tool(&self.tool_name, serde_json::to_value(arguments)?)
            .await
    }
}

/// Configuration for the MCP server
#[derive(Debug, Clone)]
pub struct ServerConfig {
    /// Maximum number of concurrent requests
    pub max_concurrent_requests: usize,
    /// Request timeout in milliseconds
    pub request_timeout_ms: u64,
    /// Whether to validate incoming requests
    pub validate_requests: bool,
    /// Whether to enable detailed logging
    pub enable_logging: bool,
}

impl Default for ServerConfig {
    fn default() -> Self {
        Self {
            max_concurrent_requests: 100,
            request_timeout_ms: 30000,
            validate_requests: true,
            enable_logging: true,
        }
    }
}

/// MCP Server implementation
///
/// The main server struct that handles MCP protocol communication, manages
/// resources, tools, and prompts, and processes client requests.
///
/// # Examples
///
/// ## Basic Server Creation
/// ```
/// use prism_mcp_rs::server::McpServer;
///
/// let server = McpServer::new("my-server".to_string(), "1.0.0".to_string());
/// ```
///
/// ## Using ServerBuilder
/// ```
/// use prism_mcp_rs::server::ServerBuilder;
///
/// let server = ServerBuilder::new()
///     .name("my-server")
///     .version("1.0.0")
///     .with_tools()
///     .with_resources()
///     .build();
/// ```
///
/// ## Adding Tools
/// ```no_run
/// use prism_mcp_rs::server::McpServer;
/// use prism_mcp_rs::core::{Tool, ToolHandler};
/// use prism_mcp_rs::core::error::McpError;
/// use serde_json::{json, Value};
/// use async_trait::async_trait;
/// use std::collections::HashMap;
///
/// struct MyTool;
///
/// #[async_trait]
/// impl ToolHandler for MyTool {
///     async fn call(&self, _args: HashMap<String, Value>)
///         -> Result<prism_mcp_rs::protocol::ToolResult, McpError>
///     {
///         Ok(prism_mcp_rs::protocol::ToolResult {
///             content: vec![],
///             is_error: Some(false),
///             structured_content: None,
///             meta: None,
///         })
///     }
/// }
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let mut server = McpServer::new("server".to_string(), "1.0.0".to_string());
///
/// // Add tool using the add_tool method directly
/// server.add_tool(
///     "my_tool",
///     Some("My custom tool"),
///     json!({}),
///     MyTool
/// ).await?;
/// # Ok(())
/// # }
/// ```
/// Main MCP server implementation
pub struct McpServer {
    /// Server information
    info: ServerInfo,
    /// Server capabilities
    capabilities: ServerCapabilities,
    /// Server configuration
    config: ServerConfig,
    /// Registered resources
    resources: Arc<RwLock<HashMap<String, Resource>>>,
    /// Registered tools
    tools: Arc<RwLock<HashMap<String, Tool>>>,
    /// Registered MCP 2026 multi-round-trip capable tools.
    multi_round_tools: Arc<RwLock<HashMap<String, RegisteredMultiRoundTool>>>,
    /// Registered tools that execute through the MCP Tasks extension.
    task_tools: Arc<RwLock<HashMap<String, RegisteredTaskTool>>>,
    /// Caller-bound durable task state.
    task_registry: TaskRegistry,
    /// Registered prompts
    prompts: Arc<RwLock<HashMap<String, Prompt>>>,
    /// Resource templates (New in 2025-11-25)
    resource_templates: Arc<RwLock<HashMap<String, ResourceTemplate>>>,
    /// Completion handlers (New in 2025-11-25)
    completion_handlers: Arc<RwLock<HashMap<String, Box<dyn CompletionHandler>>>>,
    /// Active transport
    transport: Arc<Mutex<Option<Box<dyn ServerTransport>>>>,
    /// Server state
    state: Arc<RwLock<ServerState>>,
    /// Request ID counter
    #[allow(dead_code)]
    request_counter: Arc<Mutex<u64>>,
    /// Authorization and rate-limiting policy shared by all transports.
    request_policy: RequestPolicy,
    /// Revisions this server accepts.
    protocol_mode: ProtocolMode,
    /// Plugin manager for dynamic tool loading
    #[cfg(feature = "plugin")]
    plugin_manager: Option<Arc<RwLock<PluginManager>>>,
}

struct RegisteredMultiRoundTool {
    info: ToolInfo,
    handler: Box<dyn MultiRoundToolHandler>,
}

#[derive(Clone)]
struct RegisteredTaskTool {
    info: ToolInfo,
    execution: TaskExecution,
    fallback: Option<Arc<dyn ToolHandler>>,
}

#[derive(Clone)]
enum TaskExecution {
    Direct(Arc<dyn TaskToolHandler>),
    Composed {
        preflight: Arc<dyn MultiRoundToolHandler>,
        handler: Arc<dyn ComposedTaskToolHandler>,
    },
}

/// Internal server state
#[derive(Debug, Clone, PartialEq)]
pub enum ServerState {
    /// Server is not yet initialized
    Uninitialized,
    /// Server is initializing
    Initializing,
    /// Server is running and ready to accept requests
    Running,
    /// Server is shutting down
    Stopping,
    /// Server has stopped
    Stopped,
}

fn request_target(request: &JsonRpcRequest) -> RequestTarget {
    let resource = request.params.as_ref().and_then(|params| {
        params
            .get("name")
            .or_else(|| params.get("uri"))
            .and_then(Value::as_str)
            .map(str::to_owned)
    });
    RequestTarget::new(request.method.clone(), resource)
}

fn validate_input_required_result(
    result: &InputRequiredResult,
    capabilities: &ClientCapabilities,
) -> McpResult<()> {
    if result.result_type != ResultType::InputRequired {
        return Err(McpError::Validation(
            "multi-round handler must return resultType input_required".to_string(),
        ));
    }
    if result.input_requests.is_empty() && result.request_state.is_none() {
        return Err(McpError::Validation(
            "input_required needs inputRequests or requestState".to_string(),
        ));
    }
    for input_request in result.input_requests.values() {
        let method = input_request
            .get("method")
            .and_then(Value::as_str)
            .ok_or_else(|| McpError::Validation("input request is missing method".to_string()))?;
        let required = match method {
            methods::SAMPLING_CREATE_MESSAGE if capabilities.sampling.is_none() => {
                Some(serde_json::json!({"sampling": {}}))
            }
            methods::ROOTS_LIST if capabilities.roots.is_none() => {
                Some(serde_json::json!({"roots": {}}))
            }
            methods::ELICITATION_CREATE if capabilities.elicitation.is_none() => {
                Some(serde_json::json!({"elicitation": {}}))
            }
            methods::SAMPLING_CREATE_MESSAGE
            | methods::ROOTS_LIST
            | methods::ELICITATION_CREATE => None,
            _ => {
                return Err(McpError::Validation(format!(
                    "unsupported input request method: {method}"
                )))
            }
        };
        if let Some(required) = required {
            return Err(McpError::MissingRequiredClientCapability(required));
        }
    }
    Ok(())
}

fn tasks_capability_error() -> McpError {
    McpError::MissingRequiredClientCapability(serde_json::json!({
        "extensions": { (TASKS_EXTENSION_ID): {} }
    }))
}

fn parse_multi_round_tool_call(
    raw_params: &Value,
    arguments: HashMap<String, Value>,
    modern_context: Option<&ModernRequestContext>,
) -> McpResult<MultiRoundToolCall> {
    let raw_object = raw_params.as_object().ok_or_else(|| {
        McpError::Validation("tool call parameters must be an object".to_string())
    })?;
    let input_responses = raw_object
        .get("inputResponses")
        .map(|value| {
            value
                .as_object()
                .cloned()
                .map(|values| values.into_iter().collect())
                .ok_or_else(|| McpError::Validation("inputResponses must be an object".to_string()))
        })
        .transpose()?
        .unwrap_or_default();
    let request_state = raw_object
        .get("requestState")
        .map(|value| {
            value
                .as_str()
                .map(str::to_string)
                .ok_or_else(|| McpError::Validation("requestState must be a string".to_string()))
        })
        .transpose()?;
    let (client_info, client_capabilities) = modern_context
        .map(|context| {
            (
                context.client_info.clone(),
                context.client_capabilities.clone(),
            )
        })
        .unwrap_or_else(|| (None, ClientCapabilities::default()));
    Ok(MultiRoundToolCall {
        arguments,
        input_responses,
        request_state,
        client_info,
        client_capabilities,
    })
}

impl McpServer {
    /// Create a new MCP server with the given name and version
    pub fn new(name: String, version: String) -> Self {
        Self {
            info: ServerInfo::new(name, version),
            capabilities: ServerCapabilities {
                extensions: None,
                prompts: Some(PromptsCapability {
                    list_changed: Some(true),
                }),
                resources: Some(ResourcesCapability {
                    subscribe: Some(true),
                    list_changed: Some(true),
                }),
                tools: Some(ToolsCapability {
                    list_changed: Some(true),
                }),
                sampling: None,
                logging: None,
                completions: Some(CompletionsCapability::default()),
                experimental: None,
            },
            config: ServerConfig::default(),
            resources: Arc::new(RwLock::new(HashMap::new())),
            tools: Arc::new(RwLock::new(HashMap::new())),
            multi_round_tools: Arc::new(RwLock::new(HashMap::new())),
            task_tools: Arc::new(RwLock::new(HashMap::new())),
            task_registry: TaskRegistry::default(),
            prompts: Arc::new(RwLock::new(HashMap::new())),
            resource_templates: Arc::new(RwLock::new(HashMap::new())),
            completion_handlers: Arc::new(RwLock::new(HashMap::new())),
            transport: Arc::new(Mutex::new(None)),
            state: Arc::new(RwLock::new(ServerState::Uninitialized)),
            request_counter: Arc::new(Mutex::new(0)),
            request_policy: RequestPolicy::default(),
            protocol_mode: ProtocolMode::Auto,
            #[cfg(feature = "plugin")]
            plugin_manager: None,
        }
    }

    /// Create a new MCP server with ergonomic string parameters
    pub fn create(name: impl Into<String>, version: impl Into<String>) -> Self {
        Self::new(name.into(), version.into())
    }

    /// Create a new MCP server with custom configuration
    pub fn with_config(name: String, version: String, config: ServerConfig) -> Self {
        let mut server = Self::new(name, version);
        server.config = config;
        server
    }

    /// Attach a plugin manager to the server
    #[cfg(feature = "plugin")]
    pub fn with_plugin_manager(mut self, manager: PluginManager) -> Self {
        self.plugin_manager = Some(Arc::new(RwLock::new(manager)));
        self
    }

    /// Internal method to sync plugin tools with server
    #[cfg(feature = "plugin")]
    pub async fn sync_plugin_tools(&self) -> McpResult<()> {
        if let Some(ref manager) = self.plugin_manager {
            let plugin_tools = manager.read().await.list_tools().await;

            for tool in plugin_tools {
                // Create a plugin tool wrapper
                let handler = PluginToolHandler::new(manager.clone(), tool.name.clone());

                // Convert the tool input schema to a simpler JSON value
                let schema = serde_json::json!({
                    "type": tool.input_schema.schema_type,
                    "properties": tool.input_schema.properties,
                    "required": tool.input_schema.required,
                });

                // Convert plugin tool to server tool format
                self.add_tool(tool.name.clone(), tool.description.clone(), schema, handler)
                    .await?;
            }
        }
        Ok(())
    }

    /// Set server capabilities
    pub fn set_capabilities(&mut self, capabilities: ServerCapabilities) {
        self.capabilities = capabilities;
    }

    /// Select dual-stack, modern-only, or legacy-only server behavior.
    pub fn set_protocol_mode(&mut self, mode: ProtocolMode) {
        self.protocol_mode = mode;
    }

    /// Return the configured protocol support policy.
    pub fn protocol_mode(&self) -> ProtocolMode {
        self.protocol_mode
    }

    /// Set server configuration
    pub fn set_config(&mut self, config: ServerConfig) {
        self.config = config;
    }

    /// Install authorization and rate-limiting controls for all requests.
    pub fn set_request_policy(&mut self, policy: RequestPolicy) {
        self.request_policy = policy;
    }

    /// Install authorization and rate-limiting controls using a builder style.
    pub fn with_request_policy(mut self, policy: RequestPolicy) -> Self {
        self.request_policy = policy;
        self
    }

    /// Set initial resources (used by ServerBuilder)
    pub fn set_initial_resources(&mut self, resources: HashMap<String, Resource>) {
        // This needs to be done synchronously during server construction
        // We'll use block_on for this initial setup
        let resources_arc = Arc::clone(&self.resources);
        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let mut resources_lock = resources_arc.write().await;
                *resources_lock = resources;
            });
        })
        .join()
        .unwrap();
    }

    /// Set initial tools (used by ServerBuilder)
    pub fn set_initial_tools(&mut self, tools: HashMap<String, Tool>) {
        let tools_arc = Arc::clone(&self.tools);
        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let mut tools_lock = tools_arc.write().await;
                *tools_lock = tools;
            });
        })
        .join()
        .unwrap();
    }

    /// Set initial prompts (used by ServerBuilder)
    pub fn set_initial_prompts(&mut self, prompts: HashMap<String, Prompt>) {
        let prompts_arc = Arc::clone(&self.prompts);
        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let mut prompts_lock = prompts_arc.write().await;
                *prompts_lock = prompts;
            });
        })
        .join()
        .unwrap();
    }

    /// Set initial resource templates (used by ServerBuilder)
    pub fn set_initial_resource_templates(&mut self, templates: HashMap<String, ResourceTemplate>) {
        let templates_arc = Arc::clone(&self.resource_templates);
        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let mut templates_lock = templates_arc.write().await;
                *templates_lock = templates;
            });
        })
        .join()
        .unwrap();
    }

    /// Create a builder for constructing servers
    pub fn builder() -> crate::server::builder::ServerBuilder {
        crate::server::builder::ServerBuilder::new()
    }

    /// Configure this server with capabilities using a fluent API
    pub fn with_capabilities(mut self, capabilities: ServerCapabilities) -> Self {
        self.capabilities = capabilities;
        self
    }

    /// Configure this server with a specific configuration (fluent API)
    pub fn with_server_config(mut self, config: ServerConfig) -> Self {
        self.config = config;
        self
    }

    /// Add a tool to the server with a ToolHandler implementation
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// // With a ToolHandler implementation
    /// server.add_tool("my_tool", Some("description"), json!({}), MyToolHandler).await?;
    /// ```
    pub async fn add_tool<H>(
        &self,
        name: impl Into<String>,
        description: Option<impl Into<String>>,
        input_schema: Value,
        handler: H,
    ) -> McpResult<()>
    where
        H: ToolHandler + 'static,
    {
        let name = name.into();
        let description = description.map(|d| d.into());
        let tool = Tool::new(name.clone(), description, input_schema, handler);
        self.multi_round_tools.write().await.remove(&name);
        self.task_tools.write().await.remove(&name);
        self.tools.write().await.insert(name, tool);
        Ok(())
    }

    /// Add a tool that may return MCP 2026 `input_required` results.
    pub async fn add_multi_round_tool<H>(
        &self,
        name: impl Into<String>,
        description: Option<impl Into<String>>,
        input_schema: Value,
        handler: H,
    ) -> McpResult<()>
    where
        H: MultiRoundToolHandler + 'static,
    {
        let name = name.into();
        let description = description.map(Into::into);
        let schema_object = input_schema.as_object().ok_or_else(|| {
            McpError::Validation("tool input schema must be an object".to_string())
        })?;
        let info = ToolInfo {
            name: name.clone(),
            description,
            input_schema: ToolInputSchema {
                schema_type: schema_object
                    .get("type")
                    .and_then(Value::as_str)
                    .unwrap_or("object")
                    .to_string(),
                properties: schema_object
                    .get("properties")
                    .and_then(Value::as_object)
                    .map(|properties| {
                        properties
                            .iter()
                            .map(|(key, value)| (key.clone(), value.clone()))
                            .collect()
                    }),
                required: schema_object
                    .get("required")
                    .and_then(Value::as_array)
                    .map(|required| {
                        required
                            .iter()
                            .filter_map(Value::as_str)
                            .map(str::to_string)
                            .collect()
                    }),
                additional_properties: schema_object
                    .iter()
                    .filter(|(key, _)| !matches!(key.as_str(), "type" | "properties" | "required"))
                    .map(|(key, value)| (key.clone(), value.clone()))
                    .collect(),
            },
            output_schema: None,
            annotations: None,
            title: None,
            icons: None,
            meta: None,
        };
        validate_tool_info(&info)?;

        self.tools.write().await.remove(&name);
        self.task_tools.write().await.remove(&name);
        self.multi_round_tools.write().await.insert(
            name,
            RegisteredMultiRoundTool {
                info,
                handler: Box::new(handler),
            },
        );
        self.emit_tools_list_changed().await
    }

    /// Add a continuation-aware tool with a fully specified tool definition.
    pub async fn add_multi_round_tool_detailed<H>(
        &self,
        info: ToolInfo,
        handler: H,
    ) -> McpResult<()>
    where
        H: MultiRoundToolHandler + 'static,
    {
        validate_tool_info(&info)?;
        let name = info.name.clone();
        self.tools.write().await.remove(&name);
        self.task_tools.write().await.remove(&name);
        self.multi_round_tools.write().await.insert(
            name,
            RegisteredMultiRoundTool {
                info,
                handler: Box::new(handler),
            },
        );
        self.emit_tools_list_changed().await
    }

    /// Add a tool using a closure
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// server.add_tool_with_closure("echo", Some("Echo tool"), json!({}), |args| {
    ///     Ok(vec![ContentBlock::text("Echo!")])
    /// }).await?;
    /// ```
    pub async fn add_tool_with_closure<F>(
        &self,
        name: impl Into<String>,
        description: Option<impl Into<String>>,
        input_schema: Value,
        handler: F,
    ) -> McpResult<()>
    where
        F: Fn(&HashMap<String, Value>) -> McpResult<Vec<ContentBlock>> + Send + Sync + 'static,
    {
        let name = name.into();
        let description = description.map(|d| d.into());
        self.add_tool(name, description, input_schema, ClosureWrapper(handler))
            .await
    }

    /// Add a tool using the ToolBuilder result
    pub async fn add_tool_built(&self, tool: Tool) -> McpResult<()> {
        let name = tool.info.name.clone();
        self.multi_round_tools.write().await.remove(&name);
        self.task_tools.write().await.remove(&name);
        self.tools.write().await.insert(name, tool);
        Ok(())
    }

    /// Register a tool whose execution is represented by a durable SEP-2663 task.
    pub async fn add_task_tool<H>(&self, info: ToolInfo, handler: H) -> McpResult<()>
    where
        H: TaskToolHandler + 'static,
    {
        validate_tool_info(&info)?;
        let name = info.name.clone();
        self.tools.write().await.remove(&name);
        self.multi_round_tools.write().await.remove(&name);
        self.task_tools.write().await.insert(
            name,
            RegisteredTaskTool {
                info,
                execution: TaskExecution::Direct(Arc::new(handler)),
                fallback: None,
            },
        );
        self.emit_tools_list_changed().await
    }

    /// Register a durable task tool plus the synchronous behavior used when a
    /// client has not negotiated `io.modelcontextprotocol/tasks`.
    pub async fn add_task_tool_with_fallback<H, F>(
        &self,
        info: ToolInfo,
        handler: H,
        fallback: F,
    ) -> McpResult<()>
    where
        H: TaskToolHandler + 'static,
        F: ToolHandler + 'static,
    {
        validate_tool_info(&info)?;
        let name = info.name.clone();
        self.tools.write().await.remove(&name);
        self.multi_round_tools.write().await.remove(&name);
        self.task_tools.write().await.insert(
            name,
            RegisteredTaskTool {
                info,
                execution: TaskExecution::Direct(Arc::new(handler)),
                fallback: Some(Arc::new(fallback)),
            },
        );
        self.emit_tools_list_changed().await
    }

    /// Register a required-task tool that gathers request-scoped input through
    /// MCP multi-round results before creating its durable task.
    ///
    /// The preflight handler returns `InputRequired` for intermediate rounds
    /// and `Complete` when the durable phase may begin. Its completed value is
    /// only a readiness signal; the composed handler produces the task result
    /// and receives the entire final [`MultiRoundToolCall`].
    pub async fn add_composed_task_tool<P, H>(
        &self,
        info: ToolInfo,
        preflight: P,
        handler: H,
    ) -> McpResult<()>
    where
        P: MultiRoundToolHandler + 'static,
        H: ComposedTaskToolHandler + 'static,
    {
        validate_tool_info(&info)?;
        let name = info.name.clone();
        self.tools.write().await.remove(&name);
        self.multi_round_tools.write().await.remove(&name);
        self.task_tools.write().await.insert(
            name,
            RegisteredTaskTool {
                info,
                execution: TaskExecution::Composed {
                    preflight: Arc::new(preflight),
                    handler: Arc::new(handler),
                },
                fallback: None,
            },
        );
        self.emit_tools_list_changed().await
    }
    /// Get server information
    pub fn info(&self) -> &ServerInfo {
        &self.info
    }

    /// Get server name (for compatibility with tests)
    pub fn name(&self) -> &str {
        &self.info.name
    }

    /// Get server version (for compatibility with tests)
    pub fn version(&self) -> &str {
        &self.info.version
    }

    /// Get server capabilities
    pub fn capabilities(&self) -> &ServerCapabilities {
        &self.capabilities
    }

    /// Get server configuration
    pub fn config(&self) -> &ServerConfig {
        &self.config
    }

    // ========================================================================
    // Resource Management
    // ========================================================================

    /// Add a resource to the server
    pub async fn add_resource<H>(&self, name: String, uri: String, handler: H) -> McpResult<()>
    where
        H: ResourceHandler + 'static,
    {
        let resource_info = ResourceInfo {
            uri: uri.clone(),
            name: name.clone(),
            description: None,
            mime_type: None,
            annotations: None,
            size: None,
            icons: None,
            title: None,
            meta: None,
        };

        validate_resource_info(&resource_info)?;

        let resource = Resource::new(resource_info, handler);

        {
            let mut resources = self.resources.write().await;
            resources.insert(uri, resource);
        }

        // Emit list changed notification if we have an active transport
        self.emit_resources_list_changed().await?;

        Ok(())
    }

    /// Add a resource with detailed information
    pub async fn add_resource_detailed<H>(&self, info: ResourceInfo, handler: H) -> McpResult<()>
    where
        H: ResourceHandler + 'static,
    {
        validate_resource_info(&info)?;

        let uri = info.uri.clone();
        let resource = Resource::new(info, handler);

        {
            let mut resources = self.resources.write().await;
            resources.insert(uri, resource);
        }

        self.emit_resources_list_changed().await?;

        Ok(())
    }

    /// Remove a resource from the server
    pub async fn remove_resource(&self, uri: &str) -> McpResult<bool> {
        let removed = {
            let mut resources = self.resources.write().await;
            resources.remove(uri).is_some()
        };

        if removed {
            self.emit_resources_list_changed().await?;
        }

        Ok(removed)
    }

    /// List all registered resources
    pub async fn list_resources(&self) -> McpResult<Vec<ResourceInfo>> {
        let resources = self.resources.read().await;
        let mut values: Vec<_> = resources
            .values()
            .map(|resource| resource.info.clone())
            .collect();
        values.sort_by(|left, right| left.uri.cmp(&right.uri));
        Ok(values)
    }

    /// Read a resource
    pub async fn read_resource(&self, uri: &str) -> McpResult<Vec<ResourceContents>> {
        let resources = self.resources.read().await;

        match resources.get(uri) {
            Some(resource) => {
                let params = HashMap::new(); // URL parameter extraction will be implemented in future versions
                resource.handler.read(uri, &params).await
            }
            None => Err(McpError::ResourceNotFound(uri.to_string())),
        }
    }

    // ========================================================================
    // Tool Management
    // ========================================================================

    /// Add a tool with detailed information
    pub async fn add_tool_detailed<H>(&self, info: ToolInfo, handler: H) -> McpResult<()>
    where
        H: ToolHandler + 'static,
    {
        validate_tool_info(&info)?;

        let name = info.name.clone();
        let tool = Tool::new(
            name.clone(),
            info.description.clone(),
            serde_json::to_value(&info.input_schema)?,
            handler,
        );

        {
            let mut tools = self.tools.write().await;
            self.multi_round_tools.write().await.remove(&name);
            self.task_tools.write().await.remove(&name);
            tools.insert(name, tool);
        }

        self.emit_tools_list_changed().await?;

        Ok(())
    }

    /// Remove a tool from the server
    pub async fn remove_tool(&self, name: &str) -> McpResult<bool> {
        let removed_regular = {
            let mut tools = self.tools.write().await;
            tools.remove(name).is_some()
        };
        let removed_multi_round = self.multi_round_tools.write().await.remove(name).is_some();
        let removed_task = self.task_tools.write().await.remove(name).is_some();
        let removed = removed_regular || removed_multi_round || removed_task;

        if removed {
            self.emit_tools_list_changed().await?;
        }

        Ok(removed)
    }

    /// List all registered tools
    pub async fn list_tools(&self) -> McpResult<Vec<ToolInfo>> {
        let tools = self.tools.read().await;
        let mut values: Vec<_> = tools.values().map(|tool| tool.info.clone()).collect();
        drop(tools);
        values.extend(
            self.multi_round_tools
                .read()
                .await
                .values()
                .map(|tool| tool.info.clone()),
        );
        values.extend(
            self.task_tools
                .read()
                .await
                .values()
                .map(|tool| tool.info.clone()),
        );
        values.sort_by(|left, right| left.name.cmp(&right.name));
        Ok(values)
    }

    /// Call a tool
    pub async fn call_tool(
        &self,
        name: &str,
        arguments: Option<HashMap<String, Value>>,
    ) -> McpResult<ToolResult> {
        let tools = self.tools.read().await;

        match tools.get(name) {
            Some(tool) => {
                if !tool.enabled {
                    return Err(McpError::ToolNotFound(format!("Tool '{name}' is disabled")));
                }

                let args = arguments.unwrap_or_default();
                tool.handler.call(args).await
            }
            None => Err(McpError::ToolNotFound(name.to_string())),
        }
    }

    // ========================================================================
    // Prompt Management
    // ========================================================================

    /// Add a prompt to the server
    pub async fn add_prompt<H>(&self, info: PromptInfo, handler: H) -> McpResult<()>
    where
        H: PromptHandler + 'static,
    {
        validate_prompt_info(&info)?;

        let name = info.name.clone();
        let prompt = Prompt::new(info, handler);

        {
            let mut prompts = self.prompts.write().await;
            prompts.insert(name, prompt);
        }

        self.emit_prompts_list_changed().await?;

        Ok(())
    }

    /// Remove a prompt from the server
    pub async fn remove_prompt(&self, name: &str) -> McpResult<bool> {
        let removed = {
            let mut prompts = self.prompts.write().await;
            prompts.remove(name).is_some()
        };

        if removed {
            self.emit_prompts_list_changed().await?;
        }

        Ok(removed)
    }

    /// List all registered prompts
    pub async fn list_prompts(&self) -> McpResult<Vec<PromptInfo>> {
        let prompts = self.prompts.read().await;
        let mut values: Vec<_> = prompts.values().map(|prompt| prompt.info.clone()).collect();
        values.sort_by(|left, right| left.name.cmp(&right.name));
        Ok(values)
    }

    /// Get a prompt
    pub async fn get_prompt(
        &self,
        name: &str,
        arguments: Option<HashMap<String, Value>>,
    ) -> McpResult<PromptResult> {
        let prompts = self.prompts.read().await;

        match prompts.get(name) {
            Some(prompt) => {
                let args = arguments.unwrap_or_default();
                prompt.handler.get(args).await
            }
            None => Err(McpError::PromptNotFound(name.to_string())),
        }
    }

    // ========================================================================
    // Resource Template Management (New in 2025-11-25)
    // ========================================================================

    /// Add a resource template to the server
    pub async fn add_resource_template(&self, template: ResourceTemplate) -> McpResult<()> {
        let name = template.name.clone();
        {
            let mut templates = self.resource_templates.write().await;
            templates.insert(name, template);
        }

        // Emit notification if supported
        self.emit_resource_templates_list_changed().await?;
        Ok(())
    }

    /// List resource templates
    pub async fn list_resource_templates(&self) -> McpResult<Vec<ResourceTemplate>> {
        let templates = self.resource_templates.read().await;
        let mut values: Vec<_> = templates.values().cloned().collect();
        values.sort_by(|left, right| left.uri_template.cmp(&right.uri_template));
        Ok(values)
    }

    /// Remove a resource template
    pub async fn remove_resource_template(&self, name: &str) -> McpResult<bool> {
        let removed = {
            let mut templates = self.resource_templates.write().await;
            templates.remove(name).is_some()
        };

        if removed {
            self.emit_resource_templates_list_changed().await?;
        }

        Ok(removed)
    }

    /// Emit resource templates list changed notification
    async fn emit_resource_templates_list_changed(&self) -> McpResult<()> {
        // In a real implementation, this would send a notification
        // For now, we'll just return Ok
        Ok(())
    }

    // ========================================================================
    // Completion Management (New in 2025-11-25)
    // ========================================================================

    /// Add completion handler for a specific reference type
    pub async fn add_completion_handler<H>(&self, ref_type: String, handler: H) -> McpResult<()>
    where
        H: CompletionHandler + 'static,
    {
        let mut handlers = self.completion_handlers.write().await;
        handlers.insert(ref_type, Box::new(handler));
        Ok(())
    }

    /// Remove completion handler for a reference type
    pub async fn remove_completion_handler(&self, ref_type: &str) -> McpResult<bool> {
        let mut handlers = self.completion_handlers.write().await;
        Ok(handlers.remove(ref_type).is_some())
    }

    /// Handle completion request
    pub async fn handle_completion(
        &self,
        reference: &CompletionReference,
        argument: &CompletionArgument,
        context: Option<&CompletionContext>,
    ) -> McpResult<Vec<String>> {
        let ref_type = match reference {
            CompletionReference::Prompt { .. } => "ref/prompt",
            CompletionReference::Resource { .. } => "ref/resource",
            CompletionReference::Tool { .. } => "ref/tool",
        };

        let handlers = self.completion_handlers.read().await;
        if let Some(handler) = handlers.get(ref_type) {
            handler.complete(reference, argument, context).await
        } else {
            // Return empty completions if no handler is registered
            Ok(vec![])
        }
    }

    // ========================================================================
    // Bidirectional Communication (New in 2025-11-25)
    // ========================================================================

    /// Send a request to the client (server-initiated)
    pub async fn send_client_request(
        &self,
        _request: JsonRpcRequest,
    ) -> McpResult<JsonRpcResponse> {
        let mut transport_guard = self.transport.lock().await;
        if let Some(_transport) = transport_guard.as_mut() {
            // For now, return an error - this would need transport-specific implementation
            Err(McpError::MethodNotFound(
                "Bidirectional communication not yet fully implemented".to_string(),
            ))
        } else {
            Err(McpError::Transport("Not connected".to_string()))
        }
    }

    /// Request LLM sampling from client
    pub async fn request_sampling(
        &self,
        params: CreateMessageParams,
    ) -> McpResult<CreateMessageResult> {
        let request = JsonRpcRequest::new(
            Value::from(self.next_request_id().await),
            methods::SAMPLING_CREATE_MESSAGE.to_string(),
            Some(params),
        )?;

        let response = self.send_client_request(request).await?;

        if let Some(result) = response.result {
            Ok(serde_json::from_value(result)?)
        } else {
            Err(McpError::Protocol("No result in response".to_string()))
        }
    }

    /// Request root directories from client
    pub async fn request_roots(&self) -> McpResult<ListRootsResult> {
        let request = JsonRpcRequest::new(
            Value::from(self.next_request_id().await),
            methods::ROOTS_LIST.to_string(),
            Some(ListRootsParams { meta: None }),
        )?;

        let response = self.send_client_request(request).await?;

        if let Some(result) = response.result {
            Ok(serde_json::from_value(result)?)
        } else {
            Err(McpError::Protocol("No result in response".to_string()))
        }
    }

    /// Request user input via elicitation
    pub async fn request_elicitation(&self, params: ElicitParams) -> McpResult<ElicitResult> {
        let request = JsonRpcRequest::new(
            Value::from(self.next_request_id().await),
            methods::ELICITATION_CREATE.to_string(),
            Some(params),
        )?;

        let response = self.send_client_request(request).await?;

        if let Some(result) = response.result {
            Ok(serde_json::from_value(result)?)
        } else {
            Err(McpError::Protocol("No result in response".to_string()))
        }
    }

    /// Request list of roots from the client
    ///
    /// Method allows the server to request access to specific directories
    /// or files on the client's file system.
    pub async fn request_list_roots(&self) -> McpResult<ListRootsResult> {
        let request = JsonRpcRequest::new(
            Value::from(self.next_request_id().await),
            methods::ROOTS_LIST.to_string(),
            None::<serde_json::Value>,
        )?;

        let response = self.send_client_request(request).await?;

        if let Some(result) = response.result {
            Ok(serde_json::from_value(result)?)
        } else {
            Err(McpError::Protocol("No result in response".to_string()))
        }
    }

    /// Send a notification that the roots list has changed
    ///
    /// This notification informs the client that the list of roots
    /// the server needs access to has changed.
    pub async fn notify_roots_list_changed(&self) -> McpResult<()> {
        let notification = JsonRpcNotification {
            jsonrpc: "2.0".to_string(),
            method: methods::ROOTS_LIST_CHANGED.to_string(),
            params: None,
        };

        if let Some(ref mut transport) = *self.transport.lock().await {
            transport.send_notification(notification).await
        } else {
            Err(McpError::Connection("No transport available".to_string()))
        }
    }

    /// Notify active MCP 2026 subscriptions that the tool catalog changed.
    pub async fn notify_tools_list_changed(&self) -> McpResult<()> {
        self.emit_tools_list_changed().await
    }

    /// Notify active MCP 2026 subscriptions that the prompt catalog changed.
    pub async fn notify_prompts_list_changed(&self) -> McpResult<()> {
        self.emit_prompts_list_changed().await
    }

    /// Notify active MCP 2026 subscriptions that the resource catalog changed.
    pub async fn notify_resources_list_changed(&self) -> McpResult<()> {
        self.emit_resources_list_changed().await
    }

    // ========================================================================
    // Server Lifecycle
    // ========================================================================

    /// Start the server with the given transport
    pub async fn start<T>(&mut self, mut transport: T) -> McpResult<()>
    where
        T: ServerTransport + 'static,
    {
        let mut state = self.state.write().await;

        match *state {
            ServerState::Uninitialized => {
                *state = ServerState::Initializing;
            }
            _ => return Err(McpError::Protocol("Server is already started".to_string())),
        }

        drop(state);

        transport.set_server_capabilities(self.modern_capabilities().await)?;
        transport.set_task_notifications(self.task_registry.subscribe())?;

        let tool_schemas = {
            let tools = self.tools.read().await;
            let mut schemas = tools
                .iter()
                .map(|(name, tool)| {
                    serde_json::to_value(&tool.info.input_schema)
                        .map(|schema| (name.clone(), schema))
                        .map_err(McpError::from)
                })
                .collect::<McpResult<HashMap<_, _>>>()?;
            drop(tools);
            for (name, tool) in self.multi_round_tools.read().await.iter() {
                schemas.insert(name.clone(), serde_json::to_value(&tool.info.input_schema)?);
            }
            for (name, tool) in self.task_tools.read().await.iter() {
                schemas.insert(name.clone(), serde_json::to_value(&tool.info.input_schema)?);
            }
            schemas
        };
        transport.set_tool_schemas(tool_schemas)?;

        // Create a request handler that delegates to this server
        let resources = self.resources.clone();
        let tools = self.tools.clone();
        let multi_round_tools = self.multi_round_tools.clone();
        let task_tools = self.task_tools.clone();
        let task_registry = self.task_registry.clone();
        let prompts = self.prompts.clone();
        let resource_templates = self.resource_templates.clone();
        let completion_handlers = self.completion_handlers.clone();
        let info = self.info.clone();
        let capabilities = self.capabilities.clone();
        let config = self.config.clone();
        let request_policy = self.request_policy.clone();
        let protocol_mode = self.protocol_mode;

        let request_handler: crate::transport::traits::ServerRequestHandler =
            Arc::new(move |request| {
                let resources = resources.clone();
                let tools = tools.clone();
                let multi_round_tools = multi_round_tools.clone();
                let task_tools = task_tools.clone();
                let task_registry = task_registry.clone();
                let prompts = prompts.clone();
                let resource_templates = resource_templates.clone();
                let completion_handlers = completion_handlers.clone();
                let info = info.clone();
                let capabilities = capabilities.clone();
                let config = config.clone();
                let request_policy = request_policy.clone();
                Box::pin(async move {
                    // Create a temporary server instance to handle the request
                    let temp_server = McpServer {
                        info,
                        capabilities,
                        config,
                        resources,
                        tools,
                        multi_round_tools,
                        task_tools,
                        task_registry,
                        prompts,
                        resource_templates,
                        completion_handlers,
                        transport: Arc::new(Mutex::new(None)),
                        state: Arc::new(RwLock::new(ServerState::Running)),
                        request_counter: Arc::new(Mutex::new(0)),
                        request_policy,
                        protocol_mode,
                        #[cfg(feature = "plugin")]
                        plugin_manager: None,
                    };
                    temp_server.handle_request(request).await
                })
            });

        // Set the request handler on the transport
        transport.set_request_handler(request_handler);

        // Set up the transport
        {
            let mut transport_guard = self.transport.lock().await;
            *transport_guard = Some(Box::new(transport));
        }

        // Start the transport
        {
            let mut transport_guard = self.transport.lock().await;
            if let Some(transport) = transport_guard.as_mut() {
                transport.start().await?;
            }
        }

        // Update state to running
        {
            let mut state = self.state.write().await;
            *state = ServerState::Running;
        }

        Ok(())
    }

    /// Stop the server
    pub async fn stop(&self) -> McpResult<()> {
        let mut state = self.state.write().await;

        match *state {
            ServerState::Running => {
                *state = ServerState::Stopping;
            }
            ServerState::Stopped => return Ok(()),
            _ => return Err(McpError::Protocol("Server is not running".to_string())),
        }

        drop(state);

        // Stop the transport
        {
            let mut transport_guard = self.transport.lock().await;
            if let Some(transport) = transport_guard.as_mut() {
                transport.stop().await?;
            }
        }

        // Update state to stopped
        {
            let mut state = self.state.write().await;
            *state = ServerState::Stopped;
        }

        Ok(())
    }

    /// Check if the server is running
    pub async fn is_running(&self) -> bool {
        let state = self.state.read().await;
        matches!(*state, ServerState::Running)
    }

    /// Get the current server state
    pub async fn state(&self) -> ServerState {
        let state = self.state.read().await;
        state.clone()
    }

    /// Start server with custom transport and run until interrupted
    ///
    /// This is a convenience method that:
    /// 1. Starts the server with the provided transport
    /// 2. Waits for Ctrl+C signal
    /// 3. smoothly shuts down the server
    ///
    /// # Arguments
    /// * `transport` - The transport to use for communication
    ///
    /// # Example
    /// ```rust,ignore
    /// use prism_mcp_rs::prelude::*;
    ///
    /// #[cfg(feature = "stdio")]
    /// use prism_mcp_rs::transport::stdio::StdioServerTransport;
    ///
    /// #[tokio::main]
    /// async fn main() -> McpResult<()> {
    ///     let mut server = McpServer::new("my-server".to_string(), "1.0.0".to_string());
    ///     // . add tools, resources, prompts ...
    ///
    ///     #[cfg(feature = "stdio")]
    ///     {
    ///         let transport = StdioServerTransport::new();
    ///         server.run_with_transport(transport).await?;
    ///     }
    ///     
    ///     Ok(())
    /// }
    /// ```
    pub async fn run_with_transport<T>(mut self, transport: T) -> McpResult<()>
    where
        T: ServerTransport + 'static,
    {
        self.start(transport).await?;

        tracing::info!("Server started. Press Ctrl+C to stop.");

        // Wait for shutdown signal
        tokio::signal::ctrl_c()
            .await
            .map_err(|e| McpError::internal(format!("Signal handling error: {e}")))?;

        tracing::info!("Shutdown signal received, stopping server...");
        self.stop().await
    }

    // ========================================================================
    // Request Handling
    // ========================================================================
    /// Handle an incoming JSON-RPC request
    pub async fn handle_request(&self, request: JsonRpcRequest) -> McpResult<JsonRpcResponse> {
        self.handle_request_with_context(request, RequestContext::anonymous())
            .await
    }

    /// Handle a request with transport-supplied identity and correlation data.
    ///
    /// Custom transports and authentication middleware should call this method
    /// after validating credentials and constructing a [`RequestContext`].
    pub async fn handle_request_with_context(
        &self,
        request: JsonRpcRequest,
        context: RequestContext,
    ) -> McpResult<JsonRpcResponse> {
        let target = request_target(&request);
        let method = request.method.clone();
        let principal = context.principal.id.clone();
        let request_id = context.request_id.clone();
        let transport = context.transport.clone();
        let span = tracing::info_span!(
            "mcp.request",
            request.id = %request_id,
            rpc.method = %method,
            principal.id = %principal,
            transport = %transport,
            otel.kind = "server"
        );

        async move {
            let started_at = std::time::Instant::now();
            if let Err(error) = self.request_policy.enforce(&context, &target).await {
                tracing::warn!(
                    event = "security.policy_denied",
                    error.category = error.category(),
                    error = %error,
                    "request rejected by production policy"
                );
                return Err(error);
            }

            // Validate the request if configured to do so
            let modern_context = match modern_request_context(&request) {
                Err(McpError::UnsupportedProtocolVersion { requested, .. }) => {
                    let supported = match self.protocol_mode {
                        ProtocolMode::Auto => SUPPORTED_PROTOCOL_VERSIONS
                            .iter()
                            .map(|version| (*version).to_string())
                            .collect(),
                        ProtocolMode::ModernOnly => vec![MODERN_PROTOCOL_VERSION.to_string()],
                        ProtocolMode::LegacyOnly => vec![LEGACY_PROTOCOL_VERSION.to_string()],
                    };
                    return Err(McpError::UnsupportedProtocolVersion {
                        requested,
                        supported,
                    });
                }
                result => result?,
            };
            let is_modern = modern_context.is_some();
            if request.method == methods::SERVER_DISCOVER && !is_modern {
                return Err(McpError::Validation(
                    "server/discover requires MCP 2026 request metadata".to_string(),
                ));
            }
            match self.protocol_mode {
                ProtocolMode::ModernOnly if !is_modern => {
                    return Err(McpError::UnsupportedProtocolVersion {
                        requested: LEGACY_PROTOCOL_VERSION.to_string(),
                        supported: vec![MODERN_PROTOCOL_VERSION.to_string()],
                    });
                }
                ProtocolMode::LegacyOnly if is_modern => {
                    return Err(McpError::UnsupportedProtocolVersion {
                        requested: MODERN_PROTOCOL_VERSION.to_string(),
                        supported: vec![LEGACY_PROTOCOL_VERSION.to_string()],
                    });
                }
                _ => {}
            }
            if is_modern && is_legacy_only_method(&request.method) {
                return Err(McpError::MethodNotFound(format!(
                    "{} is unavailable in MCP {MODERN_PROTOCOL_VERSION}",
                    request.method
                )));
            }

            if self.config.validate_requests {
                validate_jsonrpc_request(&request)?;
                validate_mcp_request(&request.method, request.params.as_ref())?;
            }

            // Route the request to the appropriate handler
            let result = match request.method.as_str() {
                methods::INITIALIZE => self.handle_initialize(request.params).await,
                methods::SERVER_DISCOVER => self.handle_server_discover().await,
                methods::PING => self.handle_ping().await,
                methods::TOOLS_LIST => self.handle_tools_list(request.params).await,
                methods::TOOLS_CALL => {
                    self.handle_tools_call(request.params, modern_context.as_ref(), &context)
                        .await
                }
                methods::TASKS_GET => {
                    self.handle_tasks_get(request.params, modern_context.as_ref(), &context)
                        .await
                }
                methods::TASKS_UPDATE => {
                    self.handle_tasks_update(request.params, modern_context.as_ref(), &context)
                        .await
                }
                methods::TASKS_CANCEL => {
                    self.handle_tasks_cancel(request.params, modern_context.as_ref(), &context)
                        .await
                }
                methods::RESOURCES_LIST => self.handle_resources_list(request.params).await,
                methods::RESOURCES_READ => self.handle_resources_read(request.params).await,
                methods::RESOURCES_SUBSCRIBE => {
                    self.handle_resources_subscribe(request.params).await
                }
                methods::RESOURCES_UNSUBSCRIBE => {
                    self.handle_resources_unsubscribe(request.params).await
                }
                methods::PROMPTS_LIST => self.handle_prompts_list(request.params).await,
                methods::PROMPTS_GET => self.handle_prompts_get(request.params).await,
                methods::RESOURCES_TEMPLATES_LIST => {
                    self.handle_resource_templates_list(request.params).await
                }
                methods::COMPLETION_COMPLETE => {
                    self.handle_completion_complete(request.params).await
                }
                methods::LOGGING_SET_LEVEL => self.handle_logging_set_level(request.params).await,
                methods::RPC_DISCOVER => self.handle_rpc_discover(request.params).await,
                _ => {
                    let method = &request.method;
                    Err(McpError::MethodNotFound(format!(
                        "Unknown method: {method}"
                    )))
                }
            };

            // Convert the result to a JSON-RPC response
            let response = match result {
                Ok(result_value) => {
                    let result_value = if is_modern {
                        decorate_modern_result(&request.method, result_value, &self.info)?
                    } else {
                        result_value
                    };
                    Ok(JsonRpcResponse::success(request.id, result_value)?)
                }
                Err(error @ McpError::Forbidden(_)) => Err(error),
                Err(error @ McpError::RateLimited { .. }) => Err(error),
                Err(error) => Err(error),
            };

            tracing::info!(
                event = "security.request_audit",
                outcome = if response.is_ok() { "allowed" } else { "error" },
                duration_ms = started_at.elapsed().as_millis() as u64,
                "request completed"
            );
            response
        }
        .instrument(span)
        .await
    }

    // ========================================================================
    // Individual Request Handlers
    // ========================================================================

    async fn handle_initialize(&self, params: Option<Value>) -> McpResult<Value> {
        let params: InitializeParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => {
                return Err(McpError::Validation(
                    "Missing initialize parameters".to_string(),
                ));
            }
        };

        validate_initialize_params(&params)?;
        if params.protocol_version != LEGACY_PROTOCOL_VERSION {
            return Err(McpError::UnsupportedProtocolVersion {
                requested: params.protocol_version,
                supported: vec![LEGACY_PROTOCOL_VERSION.to_string()],
            });
        }

        let result = InitializeResult::new(
            LEGACY_PROTOCOL_VERSION.to_string(),
            self.capabilities.clone(),
            self.info.clone(),
        );

        Ok(serde_json::to_value(result)?)
    }

    async fn handle_server_discover(&self) -> McpResult<Value> {
        let supported_versions: Vec<&str> = match self.protocol_mode {
            ProtocolMode::Auto => SUPPORTED_PROTOCOL_VERSIONS.to_vec(),
            ProtocolMode::ModernOnly => vec![MODERN_PROTOCOL_VERSION],
            ProtocolMode::LegacyOnly => vec![LEGACY_PROTOCOL_VERSION],
        };
        Ok(serde_json::json!({
            "supportedVersions": supported_versions,
            "capabilities": self.modern_capabilities().await,
            "instructions": self.config.enable_logging.then_some(
                "Prism MCP server with stateless 2026 and legacy 2025 interoperability"
            )
        }))
    }

    async fn modern_capabilities(&self) -> ServerCapabilities {
        let mut capabilities = self.capabilities.clone();
        if let Some(tools) = capabilities.tools.as_mut() {
            tools.list_changed = Some(true);
        }
        if let Some(prompts) = capabilities.prompts.as_mut() {
            prompts.list_changed = Some(true);
        }
        if let Some(resources) = capabilities.resources.as_mut() {
            resources.list_changed = Some(true);
        }
        if !self.task_tools.read().await.is_empty() {
            capabilities
                .extensions
                .get_or_insert_with(HashMap::new)
                .insert(TASKS_EXTENSION_ID.to_string(), serde_json::json!({}));
        }
        capabilities
    }

    async fn handle_ping(&self) -> McpResult<Value> {
        Ok(serde_json::to_value(PingResult { meta: None })?)
    }

    async fn handle_tools_list(&self, params: Option<Value>) -> McpResult<Value> {
        let _params: ListToolsParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => ListToolsParams::default(),
        };

        let tools = self.list_tools().await?;
        let result = ListToolsResult {
            tools,
            next_cursor: None, // Pagination support will be added in future versions
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    async fn handle_tools_call(
        &self,
        params: Option<Value>,
        modern_context: Option<&ModernRequestContext>,
        request_context: &RequestContext,
    ) -> McpResult<Value> {
        let raw_params = match params {
            Some(params) => params,
            None => {
                return Err(McpError::Validation(
                    "Missing tool call parameters".to_string(),
                ));
            }
        };
        let params: CallToolParams = serde_json::from_value(raw_params.clone())?;

        validate_call_tool_params(&params)?;

        let task_tool = self.task_tools.read().await.get(&params.name).cloned();
        if let Some(tool) = task_tool {
            let tasks_negotiated = modern_context
                .map(|context| has_tasks_extension(&context.client_capabilities))
                .unwrap_or(false);
            if !tasks_negotiated {
                if let Some(fallback) = &tool.fallback {
                    return Ok(serde_json::to_value(
                        fallback.call(params.arguments.unwrap_or_default()).await?,
                    )?);
                }
                return Err(tasks_capability_error());
            }
            let task = match tool.execution {
                TaskExecution::Direct(handler) => {
                    self.task_registry
                        .create(
                            request_context.principal.id.clone(),
                            params.arguments.unwrap_or_default(),
                            handler,
                        )
                        .await?
                }
                TaskExecution::Composed { preflight, handler } => {
                    let call = parse_multi_round_tool_call(
                        &raw_params,
                        params.arguments.unwrap_or_default(),
                        modern_context,
                    )?;
                    match preflight.call(call.clone()).await? {
                        OperationResult::Complete(_) => {
                            self.task_registry
                                .create_composed(
                                    request_context.principal.id.clone(),
                                    call,
                                    handler,
                                )
                                .await?
                        }
                        OperationResult::InputRequired(result) => {
                            validate_input_required_result(&result, &call.client_capabilities)?;
                            return Ok(serde_json::to_value(result)?);
                        }
                    }
                }
            };
            return Ok(serde_json::to_value(CreateTaskResult {
                result_type: "task".to_string(),
                task,
                meta: HashMap::new(),
            })?);
        }

        if let Some(tool) = self.multi_round_tools.read().await.get(&params.name) {
            let call = parse_multi_round_tool_call(
                &raw_params,
                params.arguments.unwrap_or_default(),
                modern_context,
            )?;
            let client_capabilities = call.client_capabilities.clone();
            return match tool.handler.call(call).await? {
                OperationResult::Complete(result) => Ok(serde_json::to_value(result)?),
                OperationResult::InputRequired(result) if modern_context.is_some() => {
                    validate_input_required_result(&result, &client_capabilities)?;
                    Ok(serde_json::to_value(result)?)
                }
                OperationResult::InputRequired(_) => Err(McpError::Protocol(
                    "input_required is unavailable in MCP 2025-11-25".to_string(),
                )),
            };
        }

        let result = self.call_tool(&params.name, params.arguments).await?;
        Ok(serde_json::to_value(result)?)
    }

    fn require_tasks_context<'a>(
        &self,
        modern_context: Option<&'a ModernRequestContext>,
    ) -> McpResult<&'a ClientCapabilities> {
        let capabilities = modern_context
            .map(|context| &context.client_capabilities)
            .ok_or_else(tasks_capability_error)?;
        if !has_tasks_extension(capabilities) {
            return Err(tasks_capability_error());
        }
        Ok(capabilities)
    }

    async fn handle_tasks_get(
        &self,
        params: Option<Value>,
        modern_context: Option<&ModernRequestContext>,
        context: &RequestContext,
    ) -> McpResult<Value> {
        self.require_tasks_context(modern_context)?;
        let params: GetTaskParams = serde_json::from_value(
            params.ok_or_else(|| McpError::InvalidParams("Missing taskId".to_string()))?,
        )?;
        let task = self
            .task_registry
            .get(&params.task_id, &context.principal.id)
            .await?;
        Ok(serde_json::to_value(GetTaskResult {
            result_type: "complete".to_string(),
            task,
            meta: HashMap::new(),
        })?)
    }

    async fn handle_tasks_update(
        &self,
        params: Option<Value>,
        modern_context: Option<&ModernRequestContext>,
        context: &RequestContext,
    ) -> McpResult<Value> {
        self.require_tasks_context(modern_context)?;
        let params: UpdateTaskParams = serde_json::from_value(
            params.ok_or_else(|| McpError::InvalidParams("Missing task update".to_string()))?,
        )?;
        self.task_registry
            .update(
                &params.task_id,
                &context.principal.id,
                params.input_responses,
            )
            .await?;
        Ok(serde_json::to_value(TaskAcknowledgement {
            result_type: "complete".to_string(),
            meta: HashMap::new(),
        })?)
    }

    async fn handle_tasks_cancel(
        &self,
        params: Option<Value>,
        modern_context: Option<&ModernRequestContext>,
        context: &RequestContext,
    ) -> McpResult<Value> {
        self.require_tasks_context(modern_context)?;
        let params: CancelTaskParams = serde_json::from_value(
            params.ok_or_else(|| McpError::InvalidParams("Missing taskId".to_string()))?,
        )?;
        self.task_registry
            .cancel(&params.task_id, &context.principal.id)
            .await?;
        Ok(serde_json::to_value(TaskAcknowledgement {
            result_type: "complete".to_string(),
            meta: HashMap::new(),
        })?)
    }

    async fn handle_resources_list(&self, params: Option<Value>) -> McpResult<Value> {
        let _params: ListResourcesParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => ListResourcesParams::default(),
        };

        let resources = self.list_resources().await?;
        let result = ListResourcesResult {
            resources,
            next_cursor: None, // Pagination support will be added in future versions
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    async fn handle_resources_read(&self, params: Option<Value>) -> McpResult<Value> {
        let params: ReadResourceParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => {
                return Err(McpError::Validation(
                    "Missing resource read parameters".to_string(),
                ));
            }
        };

        validate_read_resource_params(&params)?;

        let contents = self.read_resource(&params.uri).await?;
        let result = ReadResourceResult {
            contents,
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    async fn handle_resources_subscribe(&self, params: Option<Value>) -> McpResult<Value> {
        let resources = self.resources.read().await;
        let result =
            crate::server::handlers::ResourceHandler::handle_subscribe(&resources, params).await?;
        Ok(serde_json::to_value(result)?)
    }

    async fn handle_resources_unsubscribe(&self, params: Option<Value>) -> McpResult<Value> {
        let resources = self.resources.read().await;
        let result =
            crate::server::handlers::ResourceHandler::handle_unsubscribe(&resources, params)
                .await?;
        Ok(serde_json::to_value(result)?)
    }

    async fn handle_prompts_list(&self, params: Option<Value>) -> McpResult<Value> {
        let _params: ListPromptsParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => ListPromptsParams::default(),
        };

        let prompts = self.list_prompts().await?;
        let result = ListPromptsResult {
            prompts,
            next_cursor: None, // Pagination support will be added in future versions
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    async fn handle_prompts_get(&self, params: Option<Value>) -> McpResult<Value> {
        let params: GetPromptParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => {
                return Err(McpError::Validation(
                    "Missing prompt get parameters".to_string(),
                ));
            }
        };

        validate_get_prompt_params(&params)?;

        let arguments = params.arguments.map(|args| {
            args.into_iter()
                .map(|(k, v)| (k, serde_json::Value::String(v)))
                .collect()
        });
        let result = self.get_prompt(&params.name, arguments).await?;
        Ok(serde_json::to_value(result)?)
    }

    async fn handle_logging_set_level(&self, params: Option<Value>) -> McpResult<Value> {
        let _params: SetLoggingLevelParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => {
                return Err(McpError::Validation(
                    "Missing logging level parameters".to_string(),
                ));
            }
        };

        // Logging level management feature planned for future implementation
        let result = SetLoggingLevelResult { meta: None };
        Ok(serde_json::to_value(result)?)
    }

    /// Handle resource templates list request (New in 2025-11-25)
    async fn handle_resource_templates_list(&self, params: Option<Value>) -> McpResult<Value> {
        let _params: ListResourceTemplatesParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => ListResourceTemplatesParams::default(),
        };

        let templates = self.list_resource_templates().await?;
        let result = ListResourceTemplatesResult {
            resource_templates: templates,
            next_cursor: None, // Pagination support will be added in future versions
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    /// Handle completion request (New in 2025-11-25)
    async fn handle_completion_complete(&self, params: Option<Value>) -> McpResult<Value> {
        let params: CompleteParams = match params {
            Some(p) => serde_json::from_value(p)?,
            None => {
                return Err(McpError::Validation(
                    "Missing completion parameters".to_string(),
                ));
            }
        };

        let completions = self
            .handle_completion(
                &params.reference,
                &params.argument,
                None, // Context support to be added
            )
            .await?;

        let result = CompleteResult {
            completion: CompletionData {
                values: completions,
                total: None,
                has_more: None,
            },
            meta: None,
        };

        Ok(serde_json::to_value(result)?)
    }

    /// Handle RPC discovery request (Optional discovery mechanism)
    async fn handle_rpc_discover(&self, params: Option<Value>) -> McpResult<Value> {
        use crate::server::discovery_handler::DiscoveryHandler;

        let handler = DiscoveryHandler::new();
        let result = handler
            .handle(&self.info, &self.capabilities, params)
            .await?;
        Ok(serde_json::to_value(result)?)
    }

    // ========================================================================
    // Notification Helpers
    // ========================================================================

    async fn emit_resources_list_changed(&self) -> McpResult<()> {
        let notification = JsonRpcNotification::new(
            methods::RESOURCES_LIST_CHANGED.to_string(),
            Some(ResourceListChangedParams { meta: None }),
        )?;

        self.send_notification(notification).await
    }

    async fn emit_tools_list_changed(&self) -> McpResult<()> {
        let notification = JsonRpcNotification::new(
            methods::TOOLS_LIST_CHANGED.to_string(),
            Some(ToolListChangedParams { meta: None }),
        )?;

        self.send_notification(notification).await
    }

    async fn emit_prompts_list_changed(&self) -> McpResult<()> {
        let notification = JsonRpcNotification::new(
            methods::PROMPTS_LIST_CHANGED.to_string(),
            Some(PromptListChangedParams { meta: None }),
        )?;

        self.send_notification(notification).await
    }

    /// Send a notification through the transport
    async fn send_notification(&self, notification: JsonRpcNotification) -> McpResult<()> {
        let mut transport_guard = self.transport.lock().await;
        if let Some(transport) = transport_guard.as_mut() {
            transport.send_notification(notification).await?;
        }
        Ok(())
    }

    // ========================================================================
    // Utility Methods
    // ========================================================================

    async fn next_request_id(&self) -> u64 {
        let mut counter = self.request_counter.lock().await;
        *counter += 1;
        *counter
    }

    // ========================================================================
    // State Management & Convenience Methods
    // ========================================================================

    /// Check if the server has been initialized
    pub async fn is_initialized(&self) -> bool {
        let state = self.state.read().await;
        matches!(*state, ServerState::Running)
    }

    /// Check if the server is stopping or stopped
    pub async fn is_stopped(&self) -> bool {
        let state = self.state.read().await;
        matches!(*state, ServerState::Stopping | ServerState::Stopped)
    }

    /// Get the current server state (alias for existing state method)
    pub async fn get_state(&self) -> ServerState {
        self.state().await
    }

    /// Check if the server has any registered tools
    pub async fn has_tools(&self) -> bool {
        let tools = self.tools.read().await;
        !tools.is_empty()
    }

    /// Check if the server has any registered resources
    pub async fn has_resources(&self) -> bool {
        let resources = self.resources.read().await;
        !resources.is_empty()
    }

    /// Check if the server has any registered prompts
    pub async fn has_prompts(&self) -> bool {
        let prompts = self.prompts.read().await;
        !prompts.is_empty()
    }

    /// Get the number of registered tools
    pub async fn tool_count(&self) -> usize {
        let tools = self.tools.read().await;
        tools.len()
    }

    /// Get the number of registered resources
    pub async fn resource_count(&self) -> usize {
        let resources = self.resources.read().await;
        resources.len()
    }

    /// Get the number of registered prompts
    pub async fn prompt_count(&self) -> usize {
        let prompts = self.prompts.read().await;
        prompts.len()
    }

    /// Check if a specific tool is registered
    pub async fn has_tool(&self, name: &str) -> bool {
        let tools = self.tools.read().await;
        tools.contains_key(name)
    }

    /// Check if a specific resource is registered
    pub async fn has_resource(&self, uri: &str) -> bool {
        let resources = self.resources.read().await;
        resources.contains_key(uri)
    }

    /// Check if a specific prompt is registered
    pub async fn has_prompt(&self, name: &str) -> bool {
        let prompts = self.prompts.read().await;
        prompts.contains_key(name)
    }

    /// Set the server state (internal use)
    async fn set_state(&self, new_state: ServerState) {
        let mut state = self.state.write().await;
        *state = new_state;
    }

    /// Initialize the server (sets state to Running)
    pub async fn initialize(&self) -> McpResult<()> {
        self.set_state(ServerState::Running).await;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[tokio::test]
    async fn test_server_creation() {
        let server = McpServer::new("test-server".to_string(), "1.0.0".to_string());
        assert_eq!(server.info().name, "test-server");
        assert_eq!(server.info().version, "1.0.0");
        assert!(!server.is_running().await);
    }

    #[tokio::test]
    async fn test_tool_management() {
        let server = McpServer::new("test-server".to_string(), "1.0.0".to_string());

        // Add a tool
        let schema = json!({
            "type": "object",
            "properties": {
                "name": {"type": "string"}
            }
        });

        struct TestToolHandler;

        #[async_trait::async_trait]
        impl ToolHandler for TestToolHandler {
            async fn call(&self, _arguments: HashMap<String, Value>) -> McpResult<ToolResult> {
                Ok(ToolResult {
                    content: vec![Content::text("Hello from tool")],
                    is_error: None,
                    structured_content: None,
                    meta: None,
                })
            }
        }

        server
            .add_tool(
                "test_tool".to_string(),
                Some("A test tool".to_string()),
                schema,
                TestToolHandler,
            )
            .await
            .unwrap();

        // List tools
        let tools = server.list_tools().await.unwrap();
        assert_eq!(tools.len(), 1);
        assert_eq!(tools[0].name, "test_tool");

        // Call tool
        let result = server.call_tool("test_tool", None).await.unwrap();
        assert_eq!(result.content.len(), 1);
    }

    #[tokio::test]
    async fn test_initialize_request() {
        let server = McpServer::new("test-server".to_string(), "1.0.0".to_string());

        let init_params = InitializeParams::new(
            crate::protocol::LEGACY_PROTOCOL_VERSION.to_string(),
            ClientCapabilities::default(),
            ClientInfo {
                name: "test-client".to_string(),
                version: "1.0.0".to_string(),
                description: None,
                title: Some("Test Client".to_string()),
                website_url: None,
                icons: None,
            },
        );

        let request =
            JsonRpcRequest::new(json!(1), methods::INITIALIZE.to_string(), Some(init_params))
                .unwrap();

        let response = server.handle_request(request).await.unwrap();
        assert!(response.result.is_some());
    }
}