import asyncio import threading import queue import time import inspect import shlex import logging from praisonaiagents._logging import get_logger import os import re import platform from typing import Any, List, Optional, Callable, Iterable, Union from functools import wraps, partial try: from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client MCP_AVAILABLE = True except ImportError: MCP_AVAILABLE = False ClientSession = None StdioServerParameters = None stdio_client = None def _is_transport_closed(exc): """True when ``exc`` means the stdio session is gone for good.""" try: import anyio if isinstance(exc, (anyio.ClosedResourceError, anyio.BrokenResourceError, anyio.EndOfStream)): return True except ImportError: pass if isinstance(exc, (BrokenPipeError, ConnectionResetError, EOFError)): return True try: from mcp.shared.exceptions import McpError from mcp.types import CONNECTION_CLOSED if isinstance(exc, McpError) and getattr(getattr(exc, "error", None), "code", None) == CONNECTION_CLOSED: return True except ImportError: pass return False class MCPToolRunner(threading.Thread): """A dedicated thread for running MCP operations.""" def __init__(self, server_params, timeout=60): super().__init__(daemon=True) self.server_params = server_params self.queue = queue.Queue() self.initialized = threading.Event() self._init_error = None # Set once the server's transport closes; the runner stops serving and # callers get this message instead of queueing onto a dead session. self._dead = None # Startup task handle so stop() can cancel a hung initialize(). self._loop = None self._main_task = None self._stop_requested = False self.tools = [] self.resources = [] self.resource_templates = [] self.prompts = [] self.timeout = timeout self._tool_timings = {} self._timings_lock = threading.Lock() self.start() def run(self): """Main thread function that processes MCP requests. On Windows the stdio transport must spawn a child process (e.g. ``npx``) and talk to it over pipes. A non-main thread gets a ``SelectorEventLoop`` by default there, which does not support subprocesses and fails with an opaque ``fileno`` error during init (issue #5598). Use a dedicated ``ProactorEventLoop`` on Windows so subprocess stdio works; other platforms keep the standard ``asyncio.run`` behaviour. """ if platform.system() == 'Windows': loop = asyncio.ProactorEventLoop() try: asyncio.set_event_loop(loop) loop.run_until_complete(self._run_async()) finally: # Mirror asyncio.run()'s graceful teardown so outstanding async # generators and the default executor are finalized before the # loop is closed (the subprocess transport relies on this). try: loop.run_until_complete(loop.shutdown_asyncgens()) except Exception: pass try: loop.run_until_complete(loop.shutdown_default_executor()) except Exception: pass try: loop.close() finally: asyncio.set_event_loop(None) else: asyncio.run(self._run_async()) async def _run_async(self): """Async entry point for MCP operations.""" self._loop = asyncio.get_running_loop() self._main_task = asyncio.current_task() try: if self._stop_requested: raise asyncio.CancelledError() # Set up MCP session async with stdio_client(self.server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize connection await session.initialize() # Get tools tools_result = await session.list_tools() self.tools = tools_result.tools # Best-effort capability negotiation for resources and prompts. # Servers that do not advertise these capabilities simply # leave the corresponding lists empty (tools-only servers are # unaffected). await self._discover_resources_and_prompts(session) # Signal that initialization is complete self.initialized.set() # Process requests. Each request runs as its own task so # one slow call never serializes the others behind it (the # session multiplexes requests by id), and the queue is read # with a short blocking get instead of a busy 10ms poll. The # bounded get keeps the reader thread from outliving the loop. loop = asyncio.get_running_loop() tasks = set() def _next_item(): try: return self.queue.get(timeout=0.25) except queue.Empty: return queue.Empty try: while True: item = await loop.run_in_executor(None, _next_item) if item is queue.Empty: if self._dead: break # leave stdio_client() so the child is reaped continue if item is None: # Shutdown signal break if self._dead: self.queue.put(item) # answered by _fail_pending() break task = asyncio.create_task(self._handle_request(session, item)) tasks.add(task) task.add_done_callback(tasks.discard) except asyncio.CancelledError: pass finally: for task in tasks: task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) except asyncio.CancelledError: # stop() cancelled a startup that never completed; leaving the # contexts above has already terminated the stdio child. if not self.initialized.is_set(): self._init_error = "MCP initialization cancelled" self.initialized.set() except Exception as e: if self.initialized.is_set(): # Session ended after a successful init (e.g. teardown of a # crashed server); report it as a lost connection, not init. self._dead = self._dead or f"MCP server connection lost: {str(e) or type(e).__name__}" else: self._init_error = f"MCP initialization error: {str(e)}" self.initialized.set() # Ensure we don't hang finally: if self.initialized.is_set() and not self._init_error and not self._dead: self._dead = "MCP server session ended" self._fail_pending() def _fail_pending(self): """Answer requests still queued after the loop exits so callers don't wait out the timeout.""" message = self._dead or self._init_error or "MCP server session ended" while True: try: item = self.queue.get(block=False) except queue.Empty: return if item is not None: try: item[0].put((False, message), block=False) except Exception: pass async def _handle_request(self, session, item): """Run one queued request on the session and post its result.""" response_queue, kind, name, arguments, cancelled = item if cancelled.is_set(): # Caller already gave up (timed out); never execute a call it was # told did not happen. return try: if kind == "resource": coro = session.read_resource(name) elif kind == "prompt": coro = session.get_prompt(name, arguments or None) else: coro = session.call_tool(name, arguments) result = await asyncio.wait_for(coro, timeout=self.timeout) response_queue.put((True, result)) except asyncio.TimeoutError: response_queue.put((False, f"MCP {kind} call timed out after {self.timeout} seconds (server side)")) except asyncio.CancelledError: # Report the real cause when a sibling call saw the transport die. response_queue.put((False, self._dead or f"MCP {kind} call cancelled: runner shut down")) raise except Exception as e: if _is_transport_closed(e): self._dead = f"MCP server connection lost: {str(e) or type(e).__name__}" response_queue.put((False, self._dead)) return response_queue.put((False, str(e) or type(e).__name__)) async def _discover_resources_and_prompts(self, session): """Enumerate server resources, resource templates and prompts. Each call is guarded independently so a server that supports only a subset of primitives (or none) still initializes cleanly. Failures are logged at debug level and leave the corresponding list empty. """ try: resources_result = await session.list_resources() self.resources = list(getattr(resources_result, "resources", []) or []) except Exception as e: logging.debug(f"MCP list_resources unavailable: {e}") try: templates_result = await session.list_resource_templates() self.resource_templates = list( getattr(templates_result, "resourceTemplates", []) or [] ) except Exception as e: logging.debug(f"MCP list_resource_templates unavailable: {e}") try: prompts_result = await session.list_prompts() self.prompts = list(getattr(prompts_result, "prompts", []) or []) except Exception as e: logging.debug(f"MCP list_prompts unavailable: {e}") def _dispatch(self, kind, name, arguments): """Send a request to the runner thread and wait for its result. Args: kind: One of ``"tool"``, ``"resource"`` or ``"prompt"``. name: Tool name, resource URI, or prompt name. arguments: Argument mapping (tools/prompts) or ``None``. Returns: The raw MCP result object on success, or ``(False, message)`` on failure so callers can format errors consistently. """ if not self.initialized.is_set(): self.initialized.wait(timeout=self.timeout) if not self.initialized.is_set(): return (False, f"MCP initialization timed out after {self.timeout} seconds") if self._init_error: return (False, self._init_error) if self._dead: return (False, self._dead) response_queue = queue.Queue(maxsize=1) cancelled = threading.Event() self.queue.put((response_queue, kind, name, arguments, cancelled)) if self._dead: self._fail_pending() # the runner may have drained the queue before our put try: success, result = response_queue.get(timeout=self.timeout) except queue.Empty: cancelled.set() return (False, f"MCP {kind} call timed out after {self.timeout} seconds") if not success: return (False, result) return result def read_resource(self, uri): """Read an MCP resource by URI and return its normalised contents.""" from .resources import normalize_resource_result result = self._dispatch("resource", uri, None) if isinstance(result, tuple) and result and result[0] is False: return f"Error: {result[1]}" return normalize_resource_result(result) def get_prompt(self, name, arguments=None): """Fetch an MCP prompt template rendered with ``arguments``.""" from .resources import normalize_prompt_result result = self._dispatch("prompt", name, arguments or {}) if isinstance(result, tuple) and result and result[0] is False: return f"Error: {result[1]}" return normalize_prompt_result(result) def call_tool(self, tool_name, arguments): """Call an MCP tool and wait for the result.""" # Import telemetry here to avoid circular imports try: from ..telemetry.telemetry import get_telemetry telemetry = get_telemetry() except (ImportError, AttributeError): telemetry = None # Check initialization first (without timing) if not self.initialized.is_set(): self.initialized.wait(timeout=self.timeout) if not self.initialized.is_set(): # Track initialization timeout failure if telemetry: telemetry.track_tool_usage(tool_name, success=False, execution_time=0) return f"Error: MCP initialization timed out after {self.timeout} seconds" if self._init_error: if telemetry: telemetry.track_tool_usage(tool_name, success=False, execution_time=0) return f"Error: {self._init_error}" if self._dead: if telemetry: telemetry.track_tool_usage(tool_name, success=False, execution_time=0) return f"Error: {self._dead}" # Start timing after initialization check start_time = time.time() is_success = False response_queue = queue.Queue(maxsize=1) try: # Put request in queue with caller-specific response channel cancelled = threading.Event() self.queue.put((response_queue, "tool", tool_name, arguments, cancelled)) if self._dead: self._fail_pending() # the runner may have drained the queue before our put # Wait for result with timeout try: success, result = response_queue.get(timeout=self.timeout) except queue.Empty: # Mark abandoned so the runner skips it if not yet started cancelled.set() return f"Error: MCP tool call timed out after {self.timeout} seconds" if not success: return f"Error: {result}" # A tool that fails at the application level returns a normal # CallToolResult with isError=True per the MCP spec. Surface it as a # structured error so retry/circuit-breaker/hook logic treats it as a # failure instead of a successful string result (issue #5319). if getattr(result, 'isError', False): if hasattr(result, 'content') and result.content: if hasattr(result.content[0], 'text'): error_text = result.content[0].text else: error_text = str(result.content[0]) else: error_text = str(result) # Guarantee a non-empty error string: an isError result with an # empty text block would otherwise yield {"error": ""}, which # retry/circuit-breaker/hook logic can misread as a success # (issue #5319). if not error_text: error_text = f"MCP tool '{tool_name}' reported an error (isError=True)" return {"error": error_text, "mcp_error": True} # Process result if hasattr(result, 'content') and result.content: if hasattr(result.content[0], 'text'): processed_result = result.content[0].text else: processed_result = str(result.content[0]) else: processed_result = str(result) is_success = True return processed_result except Exception as e: return f"Error: {str(e)}" finally: # Track timing regardless of success/failure end_time = time.time() execution_time = end_time - start_time # Log timing information for debugging logging.debug(f"Tool '{tool_name}' execution time: {execution_time:.3f} seconds") # Store timing in thread-safe manner with self._timings_lock: self._tool_timings[tool_name] = execution_time # Track tool usage with timing information if telemetry: telemetry.track_tool_usage(tool_name, success=is_success, execution_time=execution_time) def shutdown(self): """Signal the thread to shut down. Queues a sentinel that the request loop consumes to exit cleanly. This only takes effect once initialization has completed and the loop is running; use :meth:`stop` for a best-effort teardown when init never finished (e.g. a timed-out handshake) so the daemon thread and its stdio child do not leak across construction retries (issue #4375). """ self.queue.put(None) def stop(self, timeout=None): """Best-effort teardown for the runner thread. Queues the shutdown sentinel and joins the thread so the request loop exits and its stdio subprocess is terminated instead of leaking a background thread (and its child) across the process lifetime. The request loop reads the sentinel promptly and cancels any in-flight calls, which are answered with an error instead of left waiting. Args: timeout: Seconds to wait for the thread to exit. Defaults to the runner's own operation ``timeout`` so an in-flight call has a chance to finish before we give up. The thread is a daemon, so this never blocks interpreter exit even if the join times out. """ try: self.queue.put(None) except Exception: pass # The sentinel is only read once init finishes, so cancel a startup # that is still in progress (e.g. hung in session.initialize()). self._stop_requested = True if not self.initialized.is_set() and self._main_task is not None: try: self._loop.call_soon_threadsafe(self._main_task.cancel) except RuntimeError: pass # loop already closed if timeout is None: timeout = getattr(self, "timeout", 60) if self.is_alive(): self.join(timeout=timeout) class MCP: """ Model Context Protocol (MCP) integration for PraisonAI Agents. This class provides a simple way to connect to MCP servers and use their tools within PraisonAI agents. Example: ```python from praisonaiagents import Agent from praisonaiagents.mcp import MCP # Method 1: Using command and args separately agent = Agent( instructions="You are a helpful assistant...", llm="gpt-4o-mini", tools=MCP( command="/path/to/python", args=["/path/to/app.py"] ) ) # Method 2: Using a single command string agent = Agent( instructions="You are a helpful assistant...", llm="gpt-4o-mini", tools=MCP("/path/to/python /path/to/app.py") ) # Method 3: Using an SSE endpoint agent = Agent( instructions="You are a helpful assistant...", llm="gpt-4o-mini", tools=MCP("http://localhost:8080/sse") ) agent.start("What is the stock price of Tesla?") ``` """ # Process-level registry backing skills' CapabilityValidator discovery # (issue #3307): name -> number of live MCP clients holding that name. # Counted rather than collected, because two live clients may legitimately # declare the same server name. A name is released, and stops satisfying a # skill requirement, once the last client holding it lets go (issue #5135). _active_server_names: dict = {} _active_server_names_lock = threading.Lock() # Registrations a GC finalizer could not release because the lock was busy # (possibly held by the frame the collector interrupted). ``list.append`` # is atomic; the next holder of the lock applies them. _pending_name_releases: list = [] # Default init/tool-call timeout (seconds). Used when the caller does not # pass an explicit ``timeout``. DEFAULT_TIMEOUT = 60 # Larger default applied only for cold-start package launchers (npx/uvx/bunx), # whose first run downloads the server package before the MCP handshake can # begin. On a fresh cache this routinely exceeds 60s (issue #5099), so the # 60s trap fires before the server ever prints "listening on stdio". An # explicit ``timeout`` from the caller always wins over this. COLD_START_TIMEOUT = 180 # Executables that fetch-and-run a package on first use; their cold start # is dominated by an npm/PyPI download, not by the MCP server itself. _COLD_START_LAUNCHERS = ('npx', 'uvx', 'bunx', 'pnpm', 'yarn') @classmethod def list_active_server_names(cls) -> set: """Return the set of MCP server names currently registered this run.""" with cls._active_server_names_lock: cls._drain_pending_name_releases() return {name for name, count in cls._active_server_names.items() if count > 0} @classmethod def _drain_pending_name_releases(cls) -> None: """Apply releases deferred by finalizers. Caller holds the lock.""" while cls._pending_name_releases: cls._adjust_server_name_counts(cls._pending_name_releases.pop(), -1) @classmethod def _adjust_server_name_counts(cls, names: tuple, delta: int) -> None: """Add ``delta`` to each name's live count, dropping names reaching zero. Must be called while holding :attr:`_active_server_names_lock`. """ for name in names: remaining = cls._active_server_names.get(name, 0) + delta if remaining > 0: cls._active_server_names[name] = remaining else: cls._active_server_names.pop(name, None) def _adopt_registered_names(self, names: tuple) -> None: """Make ``names`` this instance's registration, releasing any previous ones. The release, the registration and the ownership update share one critical section, so a concurrent :meth:`shutdown` cannot observe the previous names after they have been handed over and release them twice. """ with type(self)._active_server_names_lock: type(self)._drain_pending_name_releases() previous = getattr(self, "_registered_server_names", None) if previous == names: return if previous: type(self)._adjust_server_name_counts(previous, -1) if names: type(self)._adjust_server_name_counts(names, 1) self._registered_server_names = names or None def _release_registered_names(self, *, blocking: bool = True) -> None: """Release this instance's registration, decrementing it exactly once. Claiming the names and decrementing them share one critical section, so concurrent lifecycle calls on the same instance (explicit ``shutdown()``, context-manager exit, agent cleanup, ``__del__``) cannot each consume a shared client's count. ``blocking=False`` is for ``__del__``: the lock is non-reentrant and the collector may have interrupted a frame on this thread that holds it. If it is busy the claimed names are queued for the next lock holder. Claiming without the lock is safe there because nothing else can still reach an object that is being finalized. """ lock = type(self)._active_server_names_lock if not lock.acquire(blocking=blocking): names = getattr(self, "_registered_server_names", None) if names: self._registered_server_names = None type(self)._pending_name_releases.append(names) return try: type(self)._drain_pending_name_releases() names = getattr(self, "_registered_server_names", None) if not names: return self._registered_server_names = None type(self)._adjust_server_name_counts(names, -1) finally: lock.release() @classmethod def _is_cold_start_launcher(cls, cmd) -> bool: """True when ``cmd`` is a package launcher (npx/uvx/...) whose first run downloads the server before the MCP handshake starts.""" if not isinstance(cmd, str): return False # Split on both separators so a Windows-style path is handled even when # this runs on a POSIX host (and vice versa). base = re.split(r'[\\/]', cmd)[-1] # Strip Windows executable extensions (npx.cmd, npx.exe, ...). base = re.sub(r'\.(cmd|exe|bat|ps1)$', '', base, flags=re.IGNORECASE).lower() return base in cls._COLD_START_LAUNCHERS def __init__(self, command_or_string=None, args=None, *, command=None, timeout=None, debug=False, allowed_tools: Optional[List[str]] = None, disabled_tools: Optional[List[str]] = None, max_response_bytes: Optional[int] = None, **kwargs): """ Initialize the MCP connection and get tools. Args: command_or_string: Either: - The command to run the MCP server (e.g., Python path) - A complete command string (e.g., "/path/to/python /path/to/app.py") - For NPX: 'npx' command with args for smithery tools - An SSE URL (e.g., "http://localhost:8080/sse") args: Arguments to pass to the command (when command_or_string is the command) command: Alternative parameter name for backward compatibility timeout: Timeout in seconds for MCP server initialization and tool calls. Defaults to 60s, or 180s for cold-start launchers such as ``npx``/``uvx`` whose first run must download the server package before the handshake starts (issue #5099). Pass an explicit value to override either default. debug: Enable debug logging for MCP operations (default: False) allowed_tools: Include whitelist - only these tools will be available (default: None = all tools) disabled_tools: Exclude blacklist - these tools will be filtered out (default: None = no exclusions) max_response_bytes: Maximum body size for each Streamable HTTP response. **kwargs: Additional parameters for StdioServerParameters Note: If both allowed_tools and disabled_tools are specified, allowed_tools takes precedence (include wins over exclude). """ # Check if MCP is available if not MCP_AVAILABLE: raise ImportError( "MCP (Model Context Protocol) package is not installed. " "Install it with: pip install praisonaiagents[mcp]" ) # Handle backward compatibility with named parameter 'command' if command_or_string is None and command is not None: command_or_string = command if max_response_bytes is not None: if isinstance(max_response_bytes, bool) or not isinstance(max_response_bytes, int) or max_response_bytes <= 0: raise ValueError("max_response_bytes must be a positive integer") if not isinstance(command_or_string, str) or not re.match(r'^https?://', command_or_string): raise ValueError("max_response_bytes is only supported for Streamable HTTP MCP servers") # Set up logging - default to WARNING level to hide INFO messages if debug: get_logger("mcp-wrapper").setLevel(logging.DEBUG) get_logger("mcp-sse").setLevel(logging.DEBUG) get_logger("mcp.client").setLevel(logging.DEBUG) get_logger("sse").setLevel(logging.DEBUG) get_logger("mcp-server").setLevel(logging.DEBUG) get_logger("mcp-client").setLevel(logging.DEBUG) get_logger("_client").setLevel(logging.DEBUG) get_logger("httpx").setLevel(logging.DEBUG) get_logger("llm").setLevel(logging.DEBUG) else: # Set all MCP-related loggers to WARNING level by default get_logger("mcp-wrapper").setLevel(logging.WARNING) get_logger("mcp-sse").setLevel(logging.WARNING) get_logger("mcp.client").setLevel(logging.WARNING) get_logger("sse").setLevel(logging.WARNING) get_logger("mcp-server").setLevel(logging.WARNING) get_logger("mcp-client").setLevel(logging.WARNING) get_logger("_client").setLevel(logging.WARNING) get_logger("httpx").setLevel(logging.WARNING) get_logger("llm").setLevel(logging.WARNING) # Store additional parameters. ``timeout=None`` is a sentinel meaning # "use the default", which lets us tell an explicit caller value apart # from the default and only auto-extend for cold-start launchers below. self._timeout_explicit = timeout is not None # URL transports (ws/http/sse) resolve to the plain default here; the # stdio path re-resolves once the launcher command is known. self.timeout = timeout if self._timeout_explicit else self.DEFAULT_TIMEOUT timeout = self.timeout self.debug = debug self.allowed_tools = allowed_tools self.disabled_tools = disabled_tools # Optional prefix applied to tool names to avoid cross-server collisions # when multiple MCP servers are loaded together (see load_mcp_tools). self._tool_prefix: Optional[str] = None # Server names this instance added to the process-level registry via # with_tool_prefix(). Released by shutdown() so the registry shrinks # when a server goes away (issue #5135). self._registered_server_names: Optional[tuple] = None # Check if this is a WebSocket URL (ws:// or wss://) if isinstance(command_or_string, str) and re.match(r'^wss?://', command_or_string): from .mcp_websocket import WebSocketMCPClient # Extract auth token if provided auth_token = kwargs.pop('auth_token', None) self.websocket_client = WebSocketMCPClient( command_or_string, debug=debug, timeout=timeout, auth_token=auth_token, options=kwargs ) self._tools = self._apply_tool_filters(list(self.websocket_client.tools)) self.is_sse = False self.is_http_stream = False self.is_websocket = True self.is_npx = False return # Check if this is an HTTP URL if isinstance(command_or_string, str) and re.match(r'^https?://', command_or_string): # Determine transport type based on URL or kwargs. Delegate the # URL->transport classification to the shared helper so there is a # single source of truth (see mcp_transport.get_transport_type). from .mcp_transport import get_transport_type if get_transport_type(command_or_string) == "sse" and 'transport_type' not in kwargs: if max_response_bytes is not None: raise ValueError("max_response_bytes is only supported for Streamable HTTP MCP servers") # Legacy SSE URL - use SSE transport for backward compatibility from .mcp_sse import SSEMCPClient self.sse_client = SSEMCPClient(command_or_string, debug=debug, timeout=timeout) self._tools = self._apply_tool_filters(list(self.sse_client.tools)) self.is_sse = True self.is_http_stream = False self.is_npx = False return else: # Use HTTP Stream transport for all other HTTP URLs from .mcp_http_stream import HTTPStreamMCPClient # Extract transport options from kwargs transport_options = {} if 'responseMode' in kwargs: transport_options['responseMode'] = kwargs.pop('responseMode') if 'headers' in kwargs: transport_options['headers'] = kwargs.pop('headers') if 'cors' in kwargs: transport_options['cors'] = kwargs.pop('cors') if 'session' in kwargs: transport_options['session'] = kwargs.pop('session') if 'resumability' in kwargs: transport_options['resumability'] = kwargs.pop('resumability') if max_response_bytes is not None: transport_options['max_response_bytes'] = max_response_bytes self.http_stream_client = HTTPStreamMCPClient( command_or_string, debug=debug, timeout=timeout, options=transport_options ) self._tools = self._apply_tool_filters(list(self.http_stream_client.tools)) self.is_sse = False self.is_http_stream = True self.is_npx = False return # Handle the single string format for stdio client. # Shell-split the command string first, then decide based on how many # tokens it yields. This makes multi-word forms like # MCP("npx -y pkg", args=["/tmp"]) work — the first token is the # executable and the remaining tokens are prepended to any explicitly # supplied args. Using shlex handles all whitespace (spaces, tabs, # newlines) and lets an executable path that contains spaces be passed # intact by quoting it, e.g. MCP('"/opt/my app"', args=["run"]). if isinstance(command_or_string, str): if platform.system() == 'Windows': # Use shlex with posix=False for Windows to handle quotes and paths with spaces parts = shlex.split(command_or_string, posix=False) # Remove quotes from parts if present (Windows shlex keeps them) parts = [part.strip('"') for part in parts] else: parts = shlex.split(command_or_string) if len(parts) > 1: # Multi-token string: first token is the executable, the rest # are prepended to any explicitly supplied args. cmd = parts[0] arguments = parts[1:] + (args or []) else: # Single token (or a quoted spaced path): treat the whole # string as a bare executable and keep explicit args as-is. cmd = parts[0] if parts else command_or_string arguments = args or [] else: # Use the original format with a bare command and separate args cmd = command_or_string arguments = args or [] # Set up stdio client self.is_sse = False self.is_http_stream = False # Cold-start launchers (npx/uvx/bunx/...) download the server package on # first run, which on a fresh cache regularly exceeds the 60s default and # trips the init timeout before the handshake even begins (issue #5099). # When the caller did not set an explicit timeout, raise the effective # init timeout for these commands. An explicit timeout always wins. if not self._timeout_explicit and self._is_cold_start_launcher(cmd): self.timeout = self.COLD_START_TIMEOUT timeout = self.timeout # Build safe environment for stdio MCP servers # Use safe baseline + explicit env from config (B5 security policy) custom_env = kwargs.get('env', {}) env = self._build_safe_env(custom_env) kwargs['env'] = env self.server_params = StdioServerParameters( command=cmd, args=arguments, **kwargs ) self.runner = MCPToolRunner(self.server_params, timeout) # Wait for initialization. Fail closed on timeout so callers never # receive a usable-but-empty MCP object that silently degrades an # Agent into a tool-less chat (issue #4375). Tear the runner down first # so a blocked handshake does not leave a background thread and stdio # child process alive across construction retries. if not self.runner.initialized.wait(timeout=self.timeout): # Fast-fail teardown: the handshake is stuck, so join briefly and # let the daemon thread be reclaimed at exit rather than blocking # the raise for the full operation timeout. self.runner.stop(timeout=1) # shlex.join preserves argv boundaries so a copy-pasteable command is # produced even when the executable path or an argument contains # spaces or shell metacharacters. full_cmd = shlex.join([str(cmd), *[str(a) for a in arguments]]) if self._is_cold_start_launcher(cmd): hint = ( f" This looks like a first-run package download; pre-warm it " f"once with `{full_cmd}` (wait for the server to start), or " f"pass a larger timeout, e.g. MCP(..., timeout=300)." ) else: hint = ( f" Verify the command runs manually (`{full_cmd}`), or pass a " f"larger timeout, e.g. MCP(..., timeout=180)." ) raise TimeoutError( f"MCP initialization timed out after {self.timeout} seconds " f"(command={cmd!r} args={arguments!r})." f"{hint}" ) if getattr(self.runner, "_init_error", None): self.runner.stop(timeout=1) raise RuntimeError( f"MCP initialization failed: {self.runner._init_error} " f"(command={cmd!r} args={arguments!r})." ) # Automatically detect if this is an NPX command base_cmd = os.path.basename(cmd) if isinstance(cmd, str) else cmd # Check for npx with or without Windows extensions npx_variants = ['npx', 'npx.cmd', 'npx.exe'] if platform.system() == 'Windows' and isinstance(base_cmd, str): # Case-insensitive comparison on Windows self.is_npx = base_cmd.lower() in [v.lower() for v in npx_variants] else: self.is_npx = base_cmd in npx_variants # For NPX-based MCP servers, use a different approach if self.is_npx: self._function_declarations = [] self._initialize_npx_mcp_tools(cmd, arguments) else: # Generate tool functions immediately and store them self._tools = self._apply_tool_filters(self._generate_tool_functions()) # Fail closed if a stdio server initialized but produced no tools. # Returning an empty iterable here would let Agent(tools=mcp) run as a # tool-less chat and hallucinate answers (issue #4375). Distinguish the # two empty cases: # * the server itself advertised zero tools -> always fail closed, # even when a filter is configured (the filter removed nothing, so # it cannot be blamed for the empty result); versus # * a filter deliberately removed every tool -> leave untouched, as # that is an intentional configuration by the caller. server_advertised_tools = bool(getattr(self.runner, "tools", None)) filter_configured = bool(self.allowed_tools or self.disabled_tools) if not self._tools and not (filter_configured and server_advertised_tools): err = getattr(self.runner, "_init_error", None) raise RuntimeError( f"MCP server produced 0 tools (command={cmd!r} args={arguments!r}). " f"init_error={err!r}. On Windows pass command='npx.cmd' explicitly " f"or use a Python MCP server (command=sys.executable, args=[server.py])." ) def _generate_tool_functions(self) -> List[Callable]: """ Generate functions for each MCP tool. Returns: List[Callable]: Functions that can be used as tools """ if self.is_sse: return list(self.sse_client.tools) if self.is_http_stream: return list(self.http_stream_client.tools) tool_functions = [] for tool in self.runner.tools: wrapper = self._create_tool_wrapper(tool) tool_functions.append(wrapper) # Register synthetic, agent-callable tools for MCP resources and # prompts, but only when the connected server actually advertises them # (tools-only servers are unaffected — no extra tools appear). tool_functions.extend(self._generate_resource_tools()) return tool_functions def _generate_resource_tools(self) -> List[Callable]: """Build synthetic tools for resources/prompts advertised by the server. Returns an empty list when the server exposes neither, so existing tools-only configurations behave exactly as before. """ runner = getattr(self, "runner", None) if runner is None: return [] from . import resources as _res # Server tools take precedence over synthetic helpers: if a server # already exposes a tool whose name collides with a synthetic one, skip # creating the synthetic callable so the callable and its OpenAI schema # stay consistent (avoids the LLM seeing one but being able to call the # other). server_tool_names = { getattr(t, "name", None) for t in getattr(runner, "tools", []) or [] } candidates: List[Callable] = [] has_resources = bool(runner.resources or runner.resource_templates) has_prompts = bool(runner.prompts) if has_resources: def list_mcp_resources() -> str: """List the resources available from the connected MCP server.""" return _res.to_json(_res.resources_to_dicts(self._live_stdio_runner(runner).resources)) def list_mcp_resource_templates() -> str: """List the resource templates available from the MCP server.""" return _res.to_json( _res.resource_templates_to_dicts(self._live_stdio_runner(runner).resource_templates) ) def read_mcp_resource(uri: str) -> str: """Read a resource from the MCP server by its URI.""" return self._live_stdio_runner(runner).read_resource(uri) candidates.extend( [list_mcp_resources, list_mcp_resource_templates, read_mcp_resource] ) if has_prompts: def list_mcp_prompts() -> str: """List the prompt templates available from the MCP server.""" return _res.to_json(_res.prompts_to_dicts(self._live_stdio_runner(runner).prompts)) def get_mcp_prompt(name: str, arguments: dict = None) -> str: """Fetch an MCP prompt template rendered with the given arguments.""" return self._live_stdio_runner(runner).get_prompt(name, arguments or {}) candidates.extend([list_mcp_prompts, get_mcp_prompt]) synthetic: List[Callable] = [ fn for fn in candidates if fn.__name__ not in server_tool_names ] return synthetic # Respawns allowed after a stdio server dies mid-session. _MAX_STDIO_RESTARTS = 3 def _live_stdio_runner(self, fallback=None): """Return the current stdio runner, respawning it if its server died. At most ``_MAX_STDIO_RESTARTS`` respawns; a replacement that fails to start is marked dead so a later call may retry within that budget. No replacement is kept once :meth:`shutdown` has begun. """ runner = getattr(self, "runner", None) or fallback if runner is None or not isinstance(getattr(runner, "_dead", None), str): return runner lock = self.__dict__.setdefault("_restart_lock", threading.Lock()) with lock: restarts = getattr(self, "_stdio_restarts", 0) if (self.runner is runner and not getattr(self, "_closing", False) and restarts < self._MAX_STDIO_RESTARTS): self._stdio_restarts = restarts + 1 logging.warning( "MCP stdio server lost (%s); restarting (%d/%d)", runner._dead, self._stdio_restarts, self._MAX_STDIO_RESTARTS, ) runner.stop(timeout=1) new_runner = MCPToolRunner(runner.server_params, runner.timeout) if not new_runner.initialized.wait(timeout=runner.timeout) or new_runner._init_error: new_runner._dead = new_runner._init_error or "MCP server restart timed out" new_runner.stop(timeout=1) self.runner = new_runner # shutdown() sets _closing before reading self.runner, so either # it sees new_runner or we see _closing here. if getattr(self, "_closing", False): new_runner.stop(timeout=1) return self.runner or runner def _call_stdio_tool(self, tool_name, arguments): """Call a stdio tool, lazily respawning the runner if its server died. The call that observed the crash is never replayed (it may already have reached the server); the next call gets a fresh runner instead. """ runner = self._live_stdio_runner() dead = getattr(runner, "_dead", None) if isinstance(dead, str): return f"Error: {dead}" return runner.call_tool(tool_name, arguments) def _create_tool_wrapper(self, tool): """Create a wrapper function for an MCP tool.""" from .mcp_schema_utils import build_tool_signature, build_openai_tool_dict input_schema = getattr(tool, 'inputSchema', None) or {} signature = build_tool_signature(input_schema) # Positional arguments map onto the named parameters of the signature, # in declaration order. Non-identifier/keyword property names (``from``, # ``max-results``) are handled through ``**kwargs``, so they are excluded # here to keep positional mapping aligned with the visible parameters. param_names = [ name for name, param in signature.parameters.items() if param.kind is not inspect.Parameter.VAR_KEYWORD ] param_annotations = { name: param.annotation for name, param in signature.parameters.items() if param.kind is not inspect.Parameter.VAR_KEYWORD } # Create function template to be properly decorated def template_function(*args, **kwargs): return None # Create a proper function with the correct signature template_function.__signature__ = signature template_function.__annotations__ = param_annotations template_function.__name__ = tool.name template_function.__qualname__ = tool.name template_function.__doc__ = tool.description # Create the actual function using a decorator @wraps(template_function) def wrapper(*args, **kwargs): # Map positional args to parameter names all_args = {} for i, arg in enumerate(args): if i < len(param_names): all_args[param_names[i]] = arg # Add keyword args all_args.update(kwargs) # Call the tool return self._call_stdio_tool(tool.name, all_args) # Make sure the wrapper has the correct signature for inspection wrapper.__signature__ = signature # Expose the server's original schema so schema builders (e.g. # LLM._generate_tool_definition) surface every property instead of # re-deriving from the signature, which omits ``**kwargs`` names # (JSON Schema allows ``from``/``max-results`` that are not valid Python # parameter names). The schema name tracks the wrapper's current # ``__name__`` so it stays in sync after ``with_tool_prefix``. def get_schema(): """Return the OpenAI tool schema built from the server's original inputSchema.""" return build_openai_tool_dict( wrapper.__name__, tool.description, input_schema ) wrapper.get_schema = get_schema return wrapper def _initialize_npx_mcp_tools(self, cmd, arguments): """Initialize the NPX MCP tools by extracting tool definitions.""" try: # For NPX tools, we'll use the same approach as regular MCP tools # but we need to handle the initialization differently if self.debug: logging.debug(f"Initializing NPX MCP tools with command: {cmd} {' '.join(arguments)}") # Generate tool functions using the regular MCP approach self._tools = self._apply_tool_filters(self._generate_tool_functions()) if self.debug: logging.debug(f"Generated {len(self._tools)} NPX MCP tools") except Exception as e: if self.debug: logging.error(f"Failed to initialize NPX MCP tools: {e}") raise RuntimeError(f"Failed to initialize NPX MCP tools: {e}") def _apply_tool_filters(self, raw_tools: List[Any]) -> List[Any]: """ Apply tool filtering based on allowed_tools and disabled_tools. Args: raw_tools: Unfiltered tool list Returns: Filtered tool list """ if not self.allowed_tools and not self.disabled_tools: return raw_tools # Import filter functions from .mcp_utils import filter_tools_by_allowlist, filter_disabled_tools # Convert tools to the format expected by filter functions (list of dicts with 'name') # Raw tools might be function objects, so we need to extract names tool_defs = [] for tool in raw_tools: if hasattr(tool, '__name__'): tool_defs.append({"name": tool.__name__, "_tool_obj": tool}) elif isinstance(tool, dict) and 'name' in tool: tool_defs.append(tool) else: # Skip tools that don't have a clear name continue # Include wins over exclude - apply allowlist exclusively if provided if self.allowed_tools: tool_defs = filter_tools_by_allowlist(tool_defs, self.allowed_tools) elif self.disabled_tools: tool_defs = filter_disabled_tools(tool_defs, self.disabled_tools) # Convert back to original format filtered_tools = [] for tool_def in tool_defs: if "_tool_obj" in tool_def: filtered_tools.append(tool_def["_tool_obj"]) else: filtered_tools.append(tool_def) return filtered_tools def apply_tool_filters( self, allowed_tools: Optional[List[str]] = None, disabled_tools: Optional[List[str]] = None, ) -> "MCP": """Public API to (re)apply allow/deny tool filters to this instance. Lets callers (e.g. the loader) express include/exclude intent without reaching into private state. ``allowed_tools`` wins over ``disabled_tools`` when both are supplied. Args: allowed_tools: Only these tool names are kept (allowlist). disabled_tools: These tool names are removed (denylist). Returns: self, to allow fluent chaining. """ if allowed_tools is not None: self.allowed_tools = allowed_tools if disabled_tools is not None: self.disabled_tools = disabled_tools if self.allowed_tools or self.disabled_tools: self._tools = self._apply_tool_filters(self._tools) return self def _build_safe_env(self, custom_env: Optional[dict] = None) -> dict: """ Build a safe environment for stdio MCP servers. Merges only safe baseline environment variables with explicit custom ones, following the security policy outlined in AGENTS.md. Args: custom_env: Optional custom environment variables from config Returns: Safe environment dictionary """ # Safe baseline environment variables safe_baseline = { 'PATH': os.environ.get('PATH', ''), 'HOME': os.environ.get('HOME', os.path.expanduser('~')), 'USER': os.environ.get('USER', ''), 'LANG': os.environ.get('LANG', 'C.UTF-8'), 'LC_ALL': os.environ.get('LC_ALL', 'C.UTF-8'), 'PYTHONIOENCODING': 'utf-8', } # Platform-specific safe variables if platform.system() == 'Windows': safe_baseline.update({ 'SYSTEMROOT': os.environ.get('SYSTEMROOT', ''), 'COMSPEC': os.environ.get('COMSPEC', ''), 'USERNAME': os.environ.get('USERNAME', ''), 'USERPROFILE': os.environ.get('USERPROFILE', ''), 'APPDATA': os.environ.get('APPDATA', ''), 'LOCALAPPDATA': os.environ.get('LOCALAPPDATA', ''), 'TEMP': os.environ.get('TEMP', ''), 'TMP': os.environ.get('TMP', ''), }) else: safe_baseline.update({ 'TMPDIR': os.environ.get('TMPDIR', '/tmp'), 'SHELL': os.environ.get('SHELL', '/bin/sh'), }) # Start with safe baseline env = {k: v for k, v in safe_baseline.items() if v} # Only include non-empty values # Add explicit custom environment variables from config if custom_env: env.update(custom_env) return env def __iter__(self) -> Iterable[Callable]: """ Allow the MCP instance to be used directly as an iterable of tools. This makes it possible to pass the MCP instance directly to the Agent's tools parameter. """ return iter(self._tools) @staticmethod def _sanitize_prefix(prefix: str) -> str: """Sanitize a server name into a safe tool-name prefix.""" sanitized = re.sub(r"[^0-9A-Za-z_]", "_", prefix or "") return sanitized.strip("_") def _prefixed_name(self, original_name: str) -> str: """Return the namespaced tool name for an original server-side name.""" if not self._tool_prefix: return original_name return f"{self._tool_prefix}_{original_name}" def with_tool_prefix(self, prefix: str) -> "MCP": """ Namespace this server's tool names with ``_`` to avoid collisions when multiple MCP servers are loaded together. Both the callable tool names (``__name__``) and the OpenAI schema names produced by :meth:`to_openai_tool` are prefixed, while calls are still dispatched to the original server-side tool names. Args: prefix: The server name to derive the prefix from. It is sanitized to contain only ``[0-9A-Za-z_]`` characters. Raises: ValueError: If ``prefix`` sanitizes to an empty string, since an empty prefix would silently reintroduce the cross-server name collisions this method exists to prevent. Returns: self, to allow fluent chaining. """ sanitized = self._sanitize_prefix(prefix) if not sanitized: raise ValueError( f"Cannot derive a valid tool prefix from server name {prefix!r}; " "provide a name containing at least one alphanumeric or " "underscore character." ) # Both spellings are registered so a skill requirement matches whichever # it declares. dict.fromkeys collapses them when they are identical, so # one shutdown releases both. names = tuple(dict.fromkeys(n for n in (prefix, sanitized) if n)) # Re-applying the same prefix *and* the same spellings is a no-op: the # tools already carry the prefix and the names are already registered. # The spellings are compared too, because a rename that sanitizes to the # same prefix (``"my-server"`` -> ``"my server"``) still changes the # original spelling a skill requirement may name. if (getattr(self, "_tool_prefix", None) == sanitized and getattr(self, "_registered_server_names", None) == names): return self # Adopting the new names releases the previous ones in the same critical # section, so a concurrent shutdown cannot release them twice. self._adopt_registered_names(names) self._tool_prefix = sanitized # Rename already-generated callable tools. Dispatch inside each # wrapper closes over the original tool name, so only the public # __name__/__qualname__ needs updating for schema construction. for tool in getattr(self, "_tools", []) or []: if callable(tool) and hasattr(tool, "__name__"): original = getattr(tool, "__original_name__", tool.__name__) tool.__original_name__ = original new_name = f"{sanitized}_{original}" tool.__name__ = new_name try: tool.__qualname__ = new_name except (AttributeError, TypeError): pass return self def get_tools(self) -> List[Callable]: """ Get the list of tool functions from this MCP instance. This method provides explicit access to the tools list, which is useful when you need to inspect or manipulate the tools programmatically. Returns: List[Callable]: List of tool functions that can be called Example: ```python # Point the filesystem server at any directory the agent may access. # Use a platform-appropriate path (e.g. "." for the current dir); # on Windows npx may need to be invoked as "npx.cmd". mcp = MCP("npx -y @modelcontextprotocol/server-filesystem .") tools = mcp.get_tools() for tool in tools: print(f"Tool: {tool.__name__}") ``` """ return self._tools def get_resources(self) -> List[dict]: """List resources/resource-templates advertised by the MCP server. Returns an empty list for transports that do not surface resources or for tools-only servers. Returns: List[dict]: Serialised resource and resource-template descriptors. """ from . import resources as _res runner = getattr(self, "runner", None) if runner is None: return [] return ( _res.resources_to_dicts(runner.resources) + _res.resource_templates_to_dicts(runner.resource_templates) ) def get_prompts(self) -> List[dict]: """List prompt templates advertised by the MCP server. Returns an empty list for transports that do not surface prompts or for servers that expose none. Returns: List[dict]: Serialised prompt descriptors with argument hints. """ from . import resources as _res runner = getattr(self, "runner", None) if runner is None: return [] return _res.prompts_to_dicts(runner.prompts) def _fix_array_schemas(self, schema): """ Fix array schemas by adding missing 'items' attribute required by OpenAI. Thin wrapper around the shared ``fix_array_schemas`` helper, kept for backward compatibility with any external subclassing. Args: schema: The schema dictionary to fix Returns: dict: The fixed schema """ from ..llm.schema_utils import fix_array_schemas return fix_array_schemas(schema) def to_openai_tool(self): """Convert the MCP tool to an OpenAI-compatible tool definition. This method is specifically invoked by the Agent class when using provider/model format (e.g., "openai/gpt-4o-mini"). Returns: dict or list: OpenAI-compatible tool definition(s) """ if self.is_sse and hasattr(self, 'sse_client') and self.sse_client.tools: # Return all tools from SSE client return self._apply_prefix_to_openai_tools(self.sse_client.to_openai_tools()) if self.is_http_stream and hasattr(self, 'http_stream_client') and self.http_stream_client.tools: # Return all tools from HTTP Stream client return self._apply_prefix_to_openai_tools(self.http_stream_client.to_openai_tools()) # Synthetic resource/prompt tools (if the server advertises them) are # captured up front so resource-only servers still expose something. synthetic_tools = self._synthetic_resource_openai_tools() # For simplicity, we'll convert the first tool only if multiple exist # More complex implementations could handle multiple tools if not hasattr(self, 'runner') or not self.runner.tools: if synthetic_tools: return synthetic_tools logging.warning("No MCP tools available to convert to OpenAI format") return None # Convert all tools to OpenAI format openai_tools = [] for tool in self.runner.tools: # Create OpenAI tool definition parameters = {} if hasattr(tool, 'inputSchema') and tool.inputSchema: # Fix array schemas to include 'items' attribute parameters = self._fix_array_schemas(tool.inputSchema) else: # Create a minimal schema if none exists parameters = { "type": "object", "properties": {}, "required": [] } openai_tools.append({ "type": "function", "function": { "name": self._prefixed_name(tool.name), "description": tool.description if hasattr(tool, 'description') else f"Call the {tool.name} tool", "parameters": parameters }, "__praisonai_deferrable__": True # Mark MCP tools as deferrable for tool search }) openai_tools.extend(synthetic_tools) return openai_tools # Names of the synthetic resource/prompt tools generated by # _generate_resource_tools. Used to distinguish them from server tools # (which live in self.runner.tools) when building OpenAI schemas. _SYNTHETIC_TOOL_NAMES = { "list_mcp_resources", "list_mcp_resource_templates", "read_mcp_resource", "list_mcp_prompts", "get_mcp_prompt", } def _synthetic_resource_openai_tools(self): """Build OpenAI schemas for the synthetic MCP resource/prompt tools. Derives schemas from the callables' own signatures so their names stay in sync with the configured tool prefix and any applied filtering. """ server_tool_names = { getattr(t, "name", None) for t in getattr(getattr(self, "runner", None), "tools", []) or [] } schemas = [] for fn in getattr(self, "_tools", []) or []: original = getattr(fn, "__original_name__", getattr(fn, "__name__", "")) if original not in self._SYNTHETIC_TOOL_NAMES or original in server_tool_names: continue properties = {} required = [] try: sig = inspect.signature(fn) for pname, param in sig.parameters.items(): ptype = "object" if pname == "arguments" else "string" properties[pname] = {"type": ptype} if param.default is inspect.Parameter.empty: required.append(pname) except (ValueError, TypeError): pass schemas.append({ "type": "function", "function": { "name": getattr(fn, "__name__", original), "description": (fn.__doc__ or f"Call the {original} tool").strip(), "parameters": { "type": "object", "properties": properties, "required": required, }, }, "__praisonai_deferrable__": True, }) return schemas def _apply_prefix_to_openai_tools(self, openai_tools): """Apply the configured tool prefix to OpenAI schema tool names. Used for transports (SSE/HTTP/WebSocket) whose clients build their own OpenAI schemas so their names stay in sync with the prefixed callable ``__name__`` values consumed during dispatch. Returns a new list with copied dicts so the call is idempotent and never mutates the clients' internal tool representations (avoids double-prefixing if a client caches its schema dicts). """ if not self._tool_prefix or not openai_tools: return openai_tools is_list = isinstance(openai_tools, list) items = openai_tools if is_list else [openai_tools] result = [] for item in items: if isinstance(item, dict): item = dict(item) fn = item.get("function") if isinstance(fn, dict) and fn.get("name"): item["function"] = dict(fn) item["function"]["name"] = self._prefixed_name(fn["name"]) result.append(item) return result if is_list else result[0] def __enter__(self): """Context manager entry - return self for use in 'with' statements.""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit - clean up resources.""" self.shutdown() return False # Don't suppress exceptions def shutdown(self): """Explicitly shut down MCP resources. Call this method when done using the MCP instance to ensure all background threads and connections are properly cleaned up. """ self._shutdown(from_finalizer=False) def _shutdown(self, *, from_finalizer: bool) -> None: """Shared teardown. ``from_finalizer`` never blocks: it signals the stdio runner instead of joining it (a join can wait up to the operation timeout inside the garbage collector) and releases the server-name registration without waiting on its lock.""" self._closing = True # stop _live_stdio_runner from installing a replacement # Shutdown stdio runner if present if hasattr(self, 'runner') and self.runner is not None: try: self.runner.shutdown() except Exception: pass # Best effort cleanup # Join the daemon thread (and terminate its stdio subprocess) so # long-lived processes don't leak one child per MCP server. Plain # shutdown() only enqueues a sentinel; stop() also joins the thread. # Not from a finalizer: the sentinel above lets the daemon exit. if not from_finalizer: try: if hasattr(self.runner, "stop"): self.runner.stop() except Exception: pass # Best effort cleanup # Only drop the reference once the thread has actually exited so a # later shutdown() can retry the join if an in-flight call kept the # runner alive past the join timeout. The thread is a daemon, so a # surviving runner never blocks interpreter exit. if not getattr(self.runner, "is_alive", lambda: False)(): self.runner = None self._close_transport_clients(wait=not from_finalizer) # Drop this server from the process-level registry now that its # connections are gone, so skills' CapabilityValidator stops reporting # it as available (issue #5135). Runs last so the registry only shrinks # once the transports above have actually been torn down. self._release_registered_names(blocking=not from_finalizer) def _close_transport_clients(self, wait: bool = True) -> None: """Close SSE / HTTP-stream / WebSocket clients (best effort). ``wait=False`` (finalizer path) schedules the WebSocket close instead of waiting up to 5 s for it on the transport's event loop.""" # Shutdown SSE client if present if hasattr(self, 'sse_client') and self.sse_client is not None: try: if hasattr(self.sse_client, 'shutdown'): self.sse_client.shutdown() except Exception: pass # Shutdown HTTP stream client if present # (HTTPStreamMCPClient exposes close(), not shutdown()) if hasattr(self, 'http_stream_client') and self.http_stream_client is not None: try: if hasattr(self.http_stream_client, 'shutdown'): self.http_stream_client.shutdown() elif hasattr(self.http_stream_client, 'close'): self.http_stream_client.close() except Exception: pass # Shutdown WebSocket client if present # (WebSocketMCPClient exposes close(), not shutdown()) if hasattr(self, 'websocket_client') and self.websocket_client is not None: try: if hasattr(self.websocket_client, 'shutdown'): self.websocket_client.shutdown() elif hasattr(self.websocket_client, 'close'): if wait: self.websocket_client.close() else: self.websocket_client.close(wait=False) except Exception: pass def __del__(self): """Clean up resources when the object is garbage collected. Note: __del__ is called during garbage collection and may not always run. For reliable cleanup, use the context manager pattern or call shutdown() explicitly. """ try: self._shutdown(from_finalizer=True) except Exception: pass # Best effort cleanup in __del__