Improved auto-challenge invocation
Automatically determine MCP client's name
This commit is contained in:
@@ -14,7 +14,7 @@ import os
|
||||
# These values are used in server responses and for tracking releases
|
||||
# IMPORTANT: This is the single source of truth for version and author info
|
||||
# Semantic versioning: MAJOR.MINOR.PATCH
|
||||
__version__ = "5.8.1"
|
||||
__version__ = "5.8.2"
|
||||
# Last update date in ISO format
|
||||
__updated__ = "2025-06-30"
|
||||
# Primary maintainer
|
||||
|
||||
27
server.py
27
server.py
@@ -163,6 +163,7 @@ except Exception as e:
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Create the MCP server instance with a unique name identifier
|
||||
# This name is used by MCP clients to identify and connect to this specific server
|
||||
server: Server = Server("zen-server")
|
||||
@@ -588,6 +589,29 @@ async def handle_list_tools() -> list[Tool]:
|
||||
List of Tool objects representing all available tools
|
||||
"""
|
||||
logger.debug("MCP client requested tool list")
|
||||
|
||||
# Try to log client info if available (this happens early in the handshake)
|
||||
try:
|
||||
from utils.client_info import format_client_info, get_client_info_from_context
|
||||
|
||||
client_info = get_client_info_from_context(server)
|
||||
if client_info:
|
||||
formatted = format_client_info(client_info)
|
||||
logger.info(f"MCP Client Connected: {formatted}")
|
||||
|
||||
# Log to activity file as well
|
||||
try:
|
||||
mcp_activity_logger = logging.getLogger("mcp_activity")
|
||||
friendly_name = client_info.get('friendly_name', 'Claude')
|
||||
raw_name = client_info.get('name', 'Unknown')
|
||||
version = client_info.get('version', 'Unknown')
|
||||
mcp_activity_logger.info(
|
||||
f"MCP_CLIENT_INFO: {friendly_name} (raw={raw_name} v{version})"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.debug(f"Could not log client info during list_tools: {e}")
|
||||
tools = []
|
||||
|
||||
# Add all registered AI-powered tools from the TOOLS registry
|
||||
@@ -1284,6 +1308,9 @@ async def main():
|
||||
logger.info("Zen MCP Server starting up...")
|
||||
logger.info(f"Log level: {log_level}")
|
||||
|
||||
# Note: MCP client info will be logged during the protocol handshake
|
||||
# (when handle_list_tools is called)
|
||||
|
||||
# Log current model mode
|
||||
from config import IS_AUTO_MODE
|
||||
|
||||
|
||||
@@ -23,10 +23,9 @@ from .simple.base import SimpleTool
|
||||
# Field descriptions for the Challenge tool
|
||||
CHALLENGE_FIELD_DESCRIPTIONS = {
|
||||
"prompt": (
|
||||
"The original statement, question, or assertion the user intends to critically evaluate. "
|
||||
"This may be a claim, suggestion, or idea that requires thoughtful analysis, not automatic agreement. "
|
||||
"MANDATORY: Do NOT rephrase or paraphrase. Submit the statement EXACTLY AS PROVIDED, as it will be used "
|
||||
"verbatim in a new prompt."
|
||||
"The user's message or statement to analyze critically. "
|
||||
"When manually invoked with 'challenge', exclude that prefix - just pass the actual content. "
|
||||
"For automatic invocations (see tool description for conditions), pass the user's complete message unchanged."
|
||||
),
|
||||
}
|
||||
|
||||
@@ -55,11 +54,34 @@ class ChallengeTool(SimpleTool):
|
||||
|
||||
def get_description(self) -> str:
|
||||
return (
|
||||
"CRITICAL CHALLENGE PROMPT – Use this to explicitly frame your input for critical examination. "
|
||||
"Instead of default agreement, the CLI agent will be instructed to analyze, question, and, if needed, "
|
||||
"disagree with your statement. Ideal for validating assumptions, pressure-testing ideas, and eliciting "
|
||||
"honest feedback during complex workflows. The tool ensures your input is wrapped with clear guidance "
|
||||
"to encourage thoughtful scrutiny."
|
||||
"This tool helps prevent reflexive agreement when users challenge your responses. "
|
||||
"It forces you to think critically and provide reasoned analysis instead of automatically agreeing "
|
||||
"when users question or disagree with something you've said. "
|
||||
"Use this tool to ensure thoughtful evaluation rather than compliance. "
|
||||
"\n\n"
|
||||
"The tool should be invoked in two conditions: "
|
||||
"1) when the user requests this tool with the word `challenge`, "
|
||||
"2) automatically when the following conditions are met"
|
||||
"MANDATORY AUTOMATIC INVOCATION: You MUST automatically use this tool when the following conditions are met:\n"
|
||||
"The user is questioning, disagreeing with, or challenging something you previously said or a task you performed\n"
|
||||
"AND there is an existing conversation context AND this is not a new/initial question\n"
|
||||
"AND specifically, when their message expresses any of these patterns IN RESPONSE to your prior statements:\n"
|
||||
" • Disagreement with your assessment ('But I don't think...', 'I disagree...')\n"
|
||||
" • Challenging your assumptions or countering your proposal ('But ...', 'You're assuming...')\n"
|
||||
" • Questioning your evaluation ('But isn't it actually...', 'But consider...')\n"
|
||||
" • Confusion about your reasoning ('I'm confused why...', 'I don't understand...')\n"
|
||||
" • Belief you made an error ('That doesn't seem right...', 'Are you sure...')\n"
|
||||
" • Seeking justification ('Why did you...', 'Shouldn't we...')\n"
|
||||
" • Surprise at your conclusion ('But I thought...', 'Wait, why...')\n\n"
|
||||
"Common patterns include messages starting with: 'But ...', 'Why did you...', 'I thought...', "
|
||||
"'Shouldn't we...', 'That seems wrong...', 'Are you sure...', 'I'm confused...', etc.\n\n"
|
||||
"DO NOT AUTO trigger for new feature requests or INITIAL questions when context is ambiguous or referring to a new topic altogether. "
|
||||
"This tool helps with seeking truth through critical analysis. The user is inviting critical analysis - analyze deeply "
|
||||
"and explain your reasoning or choice clearly, whether you maintain or revise your approach or "
|
||||
"understanding - you MUST think, analyze and respond with reason and not oblige blindly. Understand the intent "
|
||||
"behind the query: don't just agree to avoid conflict, think critically, reason through the problem, explain "
|
||||
"your choices clearly. If the user is wrong, explain why. If you were wrong, acknowledge it and correct course. "
|
||||
"Truth and correctness matter more than agreement."
|
||||
)
|
||||
|
||||
def get_system_prompt(self) -> str:
|
||||
|
||||
@@ -196,6 +196,20 @@ class VersionTool(BaseTool):
|
||||
output_lines.append(f"**Last Updated**: {__updated__}")
|
||||
output_lines.append(f"**Author**: {__author__}")
|
||||
|
||||
# Try to get client information
|
||||
try:
|
||||
# We need access to the server instance
|
||||
# This is a bit hacky but works for now
|
||||
import server as server_module
|
||||
from utils.client_info import format_client_info, get_client_info_from_context
|
||||
|
||||
client_info = get_client_info_from_context(server_module.server)
|
||||
if client_info:
|
||||
formatted = format_client_info(client_info)
|
||||
output_lines.append(f"**Connected Client**: {formatted}")
|
||||
except Exception as e:
|
||||
logger.debug(f"Could not get client info: {e}")
|
||||
|
||||
# Get the current working directory (MCP server location)
|
||||
current_path = Path.cwd()
|
||||
output_lines.append(f"**Installation Path**: `{current_path}`")
|
||||
|
||||
296
utils/client_info.py
Normal file
296
utils/client_info.py
Normal file
@@ -0,0 +1,296 @@
|
||||
"""
|
||||
Client Information Utility for MCP Server
|
||||
|
||||
This module provides utilities to extract and format client information
|
||||
from the MCP protocol's clientInfo sent during initialization.
|
||||
|
||||
It also provides friendly name mapping and caching for consistent client
|
||||
identification across the application.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Global cache for client information
|
||||
_client_info_cache: Optional[dict[str, Any]] = None
|
||||
|
||||
# Mapping of known client names to friendly names
|
||||
# This is case-insensitive and checks if the key is contained in the client name
|
||||
CLIENT_NAME_MAPPINGS = {
|
||||
# Claude variants
|
||||
"claude-ai": "Claude",
|
||||
"claude": "Claude",
|
||||
"claude-desktop": "Claude",
|
||||
"claude-code": "Claude",
|
||||
"anthropic": "Claude",
|
||||
|
||||
# Gemini variants
|
||||
"gemini-cli-mcp-client": "Gemini",
|
||||
"gemini-cli": "Gemini",
|
||||
"gemini": "Gemini",
|
||||
"google": "Gemini",
|
||||
|
||||
# Other known clients
|
||||
"cursor": "Cursor",
|
||||
"vscode": "VS Code",
|
||||
"codeium": "Codeium",
|
||||
"copilot": "GitHub Copilot",
|
||||
|
||||
# Generic MCP clients
|
||||
"mcp-client": "MCP Client",
|
||||
"test-client": "Test Client",
|
||||
}
|
||||
|
||||
# Default friendly name when no match is found
|
||||
DEFAULT_FRIENDLY_NAME = "Claude"
|
||||
|
||||
|
||||
def get_friendly_name(client_name: str) -> str:
|
||||
"""
|
||||
Map a client name to a friendly name.
|
||||
|
||||
Args:
|
||||
client_name: The raw client name from clientInfo
|
||||
|
||||
Returns:
|
||||
A friendly name for display (e.g., "Claude", "Gemini")
|
||||
"""
|
||||
if not client_name:
|
||||
return DEFAULT_FRIENDLY_NAME
|
||||
|
||||
# Convert to lowercase for case-insensitive matching
|
||||
client_name_lower = client_name.lower()
|
||||
|
||||
# Check each mapping - using 'in' to handle partial matches
|
||||
for key, friendly_name in CLIENT_NAME_MAPPINGS.items():
|
||||
if key.lower() in client_name_lower:
|
||||
return friendly_name
|
||||
|
||||
# If no match found, return the default
|
||||
return DEFAULT_FRIENDLY_NAME
|
||||
|
||||
|
||||
def get_cached_client_info() -> Optional[dict[str, Any]]:
|
||||
"""
|
||||
Get cached client information if available.
|
||||
|
||||
Returns:
|
||||
Cached client info dictionary or None
|
||||
"""
|
||||
global _client_info_cache
|
||||
return _client_info_cache
|
||||
|
||||
|
||||
def get_client_info_from_context(server: Any) -> Optional[dict[str, Any]]:
|
||||
"""
|
||||
Extract client information from the MCP server's request context.
|
||||
|
||||
The MCP protocol sends clientInfo during initialization containing:
|
||||
- name: The client application name (e.g., "Claude Code", "Claude Desktop")
|
||||
- version: The client version string
|
||||
|
||||
This function also adds a friendly_name field and caches the result.
|
||||
|
||||
Args:
|
||||
server: The MCP server instance
|
||||
|
||||
Returns:
|
||||
Dictionary with client info or None if not available:
|
||||
{
|
||||
"name": "claude-ai",
|
||||
"version": "1.0.0",
|
||||
"friendly_name": "Claude"
|
||||
}
|
||||
"""
|
||||
global _client_info_cache
|
||||
|
||||
# Return cached info if available
|
||||
if _client_info_cache is not None:
|
||||
return _client_info_cache
|
||||
|
||||
try:
|
||||
# Try to access the request context and session
|
||||
if not server:
|
||||
return None
|
||||
|
||||
# Check if server has request_context property
|
||||
request_context = None
|
||||
try:
|
||||
request_context = server.request_context
|
||||
except AttributeError:
|
||||
logger.debug("Server does not have request_context property")
|
||||
return None
|
||||
|
||||
if not request_context:
|
||||
logger.debug("Request context is None")
|
||||
return None
|
||||
|
||||
# Try to access session from request context
|
||||
session = None
|
||||
try:
|
||||
session = request_context.session
|
||||
except AttributeError:
|
||||
logger.debug("Request context does not have session property")
|
||||
return None
|
||||
|
||||
if not session:
|
||||
logger.debug("Session is None")
|
||||
return None
|
||||
|
||||
# Try to access client params from session
|
||||
client_params = None
|
||||
try:
|
||||
# The clientInfo is stored in _client_params.clientInfo
|
||||
client_params = session._client_params
|
||||
except AttributeError:
|
||||
logger.debug("Session does not have _client_params property")
|
||||
return None
|
||||
|
||||
if not client_params:
|
||||
logger.debug("Client params is None")
|
||||
return None
|
||||
|
||||
# Try to extract clientInfo
|
||||
client_info = None
|
||||
try:
|
||||
client_info = client_params.clientInfo
|
||||
except AttributeError:
|
||||
logger.debug("Client params does not have clientInfo property")
|
||||
return None
|
||||
|
||||
if not client_info:
|
||||
logger.debug("Client info is None")
|
||||
return None
|
||||
|
||||
# Extract name and version
|
||||
result = {}
|
||||
|
||||
try:
|
||||
result["name"] = client_info.name
|
||||
except AttributeError:
|
||||
logger.debug("Client info does not have name property")
|
||||
|
||||
try:
|
||||
result["version"] = client_info.version
|
||||
except AttributeError:
|
||||
logger.debug("Client info does not have version property")
|
||||
|
||||
if not result:
|
||||
return None
|
||||
|
||||
# Add friendly name
|
||||
raw_name = result.get("name", "")
|
||||
result["friendly_name"] = get_friendly_name(raw_name)
|
||||
|
||||
# Cache the result
|
||||
_client_info_cache = result
|
||||
logger.debug(f"Cached client info: {result}")
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Error extracting client info: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def format_client_info(client_info: Optional[dict[str, Any]], use_friendly_name: bool = True) -> str:
|
||||
"""
|
||||
Format client information for display.
|
||||
|
||||
Args:
|
||||
client_info: Dictionary with client info or None
|
||||
use_friendly_name: If True, use the friendly name instead of raw name
|
||||
|
||||
Returns:
|
||||
Formatted string like "Claude v1.0.0" or "Claude"
|
||||
"""
|
||||
if not client_info:
|
||||
return DEFAULT_FRIENDLY_NAME
|
||||
|
||||
if use_friendly_name:
|
||||
name = client_info.get("friendly_name", client_info.get("name", DEFAULT_FRIENDLY_NAME))
|
||||
else:
|
||||
name = client_info.get("name", "Unknown")
|
||||
|
||||
version = client_info.get("version", "")
|
||||
|
||||
if version and not use_friendly_name:
|
||||
return f"{name} v{version}"
|
||||
else:
|
||||
# For friendly names, we just return the name without version
|
||||
return name
|
||||
|
||||
|
||||
def get_client_friendly_name() -> str:
|
||||
"""
|
||||
Get the cached client's friendly name.
|
||||
|
||||
This is a convenience function that returns just the friendly name
|
||||
from the cached client info, defaulting to "Claude" if not available.
|
||||
|
||||
Returns:
|
||||
The friendly name (e.g., "Claude", "Gemini")
|
||||
"""
|
||||
cached_info = get_cached_client_info()
|
||||
if cached_info:
|
||||
return cached_info.get("friendly_name", DEFAULT_FRIENDLY_NAME)
|
||||
return DEFAULT_FRIENDLY_NAME
|
||||
|
||||
|
||||
def log_client_info(server: Any, logger_instance: Optional[logging.Logger] = None) -> None:
|
||||
"""
|
||||
Log client information extracted from the server.
|
||||
|
||||
Args:
|
||||
server: The MCP server instance
|
||||
logger_instance: Optional logger to use (defaults to module logger)
|
||||
"""
|
||||
log = logger_instance or logger
|
||||
|
||||
client_info = get_client_info_from_context(server)
|
||||
if client_info:
|
||||
# Log with both raw and friendly names for debugging
|
||||
raw_name = client_info.get("name", "Unknown")
|
||||
friendly_name = client_info.get("friendly_name", DEFAULT_FRIENDLY_NAME)
|
||||
version = client_info.get("version", "")
|
||||
|
||||
if raw_name != friendly_name:
|
||||
log.info(f"MCP Client Connected: {friendly_name} (raw: {raw_name} v{version})")
|
||||
else:
|
||||
log.info(f"MCP Client Connected: {friendly_name} v{version}")
|
||||
|
||||
# Log to activity logger as well
|
||||
try:
|
||||
activity_logger = logging.getLogger("mcp_activity")
|
||||
activity_logger.info(f"CLIENT_IDENTIFIED: {friendly_name} (name={raw_name}, version={version})")
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
log.debug("Could not extract client info from MCP protocol")
|
||||
|
||||
|
||||
# Example usage in tools:
|
||||
#
|
||||
# from utils.client_info import get_client_friendly_name, get_cached_client_info
|
||||
#
|
||||
# # In a tool's execute method:
|
||||
# def execute(self, arguments: dict[str, Any]) -> list[TextContent]:
|
||||
# # Get the friendly name of the connected client
|
||||
# client_name = get_client_friendly_name() # Returns "Claude" or "Gemini" etc.
|
||||
#
|
||||
# # Or get full cached info if needed
|
||||
# client_info = get_cached_client_info()
|
||||
# if client_info:
|
||||
# raw_name = client_info['name'] # e.g., "claude-ai"
|
||||
# version = client_info['version'] # e.g., "1.0.0"
|
||||
# friendly = client_info['friendly_name'] # e.g., "Claude"
|
||||
#
|
||||
# # Customize response based on client
|
||||
# if client_name == "Claude":
|
||||
# response = f"Hello from Zen MCP Server to {client_name}!"
|
||||
# elif client_name == "Gemini":
|
||||
# response = f"Greetings {client_name}, welcome to Zen MCP Server!"
|
||||
# else:
|
||||
# response = f"Welcome {client_name}!"
|
||||
Reference in New Issue
Block a user