MCP Client 的工作原理：构建智能工具调用的核心引擎

class MCParser:
    def __init__(self, config=None):
        self.config = config or {}
        self.supported_formats = self.config.get('supported_formats', ['json', 'yaml'])
        self.default_format = self.config.get('default_format', 'json')
    
    def parse(self, model_output):
        """解析模型输出，识别工具调用请求"""
        # 1. 预处理模型输出
        processed_output = self._preprocess_output(model_output)
        
        # 2. 检测输出格式
        output_format = self._detect_format(processed_output)
        
        # 3. 解析输出内容
        parsed_content = self._parse_content(processed_output, output_format)
        
        # 4. 识别工具调用请求
        tool_calls = self._extract_tool_calls(parsed_content)
        
        return {
            'format': output_format,
            'parsed_content': parsed_content,
            'tool_calls': tool_calls
        }
    
    def _preprocess_output(self, output):
        """预处理模型输出，去除无关内容"""
        # 移除代码块标记
        if output.startswith('```'):
            output = output.split('```')[1]
            if ':' in output:  # 处理带语言标记的代码块
                output = output.split(':', 1)[1].strip()
        
        # 移除首尾空白字符
        return output.strip()
    
    def _detect_format(self, output):
        """检测输出格式"""
        # 简单的格式检测逻辑
        if output.startswith('{') or output.endswith('}'):
            return 'json'
        elif output.startswith('---') or ':' in output:
            return 'yaml'
        return self.default_format
    
    def _parse_content(self, output, format_type):
        """根据格式类型解析输出内容"""
        if format_type == 'json':
            import json
            return json.loads(output)
        elif format_type == 'yaml':
            import yaml
            return yaml.safe_load(output)
        raise ValueError(f"Unsupported format: {format_type}")
    
    def _extract_tool_calls(self, content):
        """从解析内容中提取工具调用请求"""
        tool_calls = []
        
        # 支持多种工具调用格式
        if isinstance(content, dict):
            # 单个工具调用
            if 'tool_call' in content:
                tool_call = content['tool_call']
                tool_calls.append({
                    'tool_name': tool_call.get('name'),
                    'params': tool_call.get('params', {}),
                    'correlation_id': tool_call.get('correlation_id')
                })
            # 多个工具调用
            elif 'tool_calls' in content:
                for tool_call in content['tool_calls']:
                    tool_calls.append({
                        'tool_name': tool_call.get('name'),
                        'params': tool_call.get('params', {}),
                        'correlation_id': tool_call.get('correlation_id')
                    })
        
        return tool_calls

class MCExecutor:
    def __init__(self, config):
        self.config = config
        self.server_url = config.get('server_url', 'http://localhost:8080')
        self.websocket_url = config.get('websocket_url', 'ws://localhost:8081')
        self.timeout = config.get('timeout', 30)
        self.max_retries = config.get('max_retries', 3)
    
    async def execute_tool(self, tool_call):
        """执行单个工具调用"""
        tool_name = tool_call['tool_name']
        params = tool_call['params']
        correlation_id = tool_call.get('correlation_id', str(uuid.uuid4()))
        
        # 选择通信方式
        if self.config.get('use_websocket', False):
            return await self._execute_via_websocket(tool_name, params, correlation_id)
        else:
            return await self._execute_via_http(tool_name, params, correlation_id)
    
    async def _execute_via_http(self, tool_name, params, correlation_id):
        """通过 HTTP 执行工具调用"""
        import httpx
        
        url = f"{self.server_url}/api/v2/tools/{tool_name}/execute"
        headers = {
            'Content-Type': 'application/json',
            'X-Correlation-ID': correlation_id
        }
        
        for attempt in range(self.max_retries):
            try:
                async with httpx.AsyncClient() as client:
                    response = await client.post(
                        url,
                        json={'params': params},
                        headers=headers,
                        timeout=self.timeout
                    )
                    response.raise_for_status()
                    return response.json()
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(1 * (2 ** attempt))  # 指数退避
    
    async def _execute_via_websocket(self, tool_name, params, correlation_id):
        """通过 WebSocket 执行工具调用"""
        import websockets
        import json
        
        async with websockets.connect(self.websocket_url) as websocket:
            # 发送工具调用请求
            await websocket.send(json.dumps({
                'type': 'tool_call',
                'tool_name': tool_name,
                'params': params,
                'correlation_id': correlation_id
            }))
            
            # 等待工具执行结果
            while True:
                response = await websocket.recv()
                result = json.loads(response)
                
                if result.get('correlation_id') == correlation_id:
                    return result
    
    async def execute_tools(self, tool_calls):
        """执行多个工具调用"""
        # 支持并行执行
        if self.config.get('parallel_execution', False):
            tasks = [self.execute_tool(tool_call) for tool_call in tool_calls]
            return await asyncio.gather(*tasks)
        # 串行执行
        else:
            results = []
            for tool_call in tool_calls:
                result = await self.execute_tool(tool_call)
                results.append(result)
            return results

class MCContextManager:
    def __init__(self, config=None):
        self.config = config or {}
        self.max_context_length = self.config.get('max_context_length', 10000)
        self.context_history = []
    
    def add_to_context(self, item):
        """添加项目到上下文"""
        self.context_history.append({
            'timestamp': datetime.now().isoformat(),
            'content': item
        })
        
        # 检查上下文长度，超过限制则裁剪
        self._trim_context()
    
    def get_context(self, max_length=None):
        """获取上下文"""
        max_length = max_length or self.max_context_length
        context_items = []
        current_length = 0
        
        # 从最新到最旧添加上下文项目
        for item in reversed(self.context_history):
            item_length = self._get_item_length(item['content'])
            if current_length + item_length <= max_length:
                context_items.insert(0, item['content'])
                current_length += item_length
            else:
                break
        
        return context_items
    
    def _trim_context(self):
        """裁剪上下文，保持在最大长度以内"""
        total_length = sum(self._get_item_length(item['content']) for item in self.context_history)
        
        while total_length > self.max_context_length and self.context_history:
            removed_item = self.context_history.pop(0)
            total_length -= self._get_item_length(removed_item['content'])
    
    def _get_item_length(self, item):
        """计算上下文项目的长度"""
        if isinstance(item, str):
            return len(item)
        elif isinstance(item, dict):
            import json
            return len(json.dumps(item))
        return len(str(item))
    
    def clear_context(self):
        """清空上下文"""
        self.context_history.clear()
    
    def get_context_summary(self):
        """获取上下文摘要"""
        return {
            'total_items': len(self.context_history),
            'total_length': sum(self._get_item_length(item['content']) for item in self.context_history),
            'first_item_timestamp': self.context_history[0]['timestamp'] if self.context_history else None,
            'last_item_timestamp': self.context_history[-1]['timestamp'] if self.context_history else None
        }

class MCClient:
    def __init__(self, config=None):
        self.config = config or {}
        self._initialize_components()
    
    def _initialize_components(self):
        """初始化各个组件"""
        from .parser import MCParser
        from .executor import MCExecutor
        from .context_manager import MCContextManager
        from .error_handler import MCErrorHandler
        from .logger import MCLogger
        from .metrics import MCMetrics
        
        self.parser = MCParser(self.config.get('parser', {}))
        self.executor = MCExecutor(self.config.get('executor', {}))
        self.context_manager = MCContextManager(self.config.get('context', {}))
        self.error_handler = MCErrorHandler(self.config.get('error', {}))
        self.logger = MCLogger(self.config.get('logging', {}))
        self.metrics = MCMetrics(self.config.get('metrics', {}))
    
    async def start(self):
        """启动 Client"""
        self.logger.info('Starting MCP Client', config=self.config)
        self.metrics.increment('client_started')
    
    async def stop(self):
        """停止 Client"""
        self.logger.info('Stopping MCP Client')
        self.metrics.increment('client_stopped')

async def process_model_output(self, model_output):
    """处理模型输出"""
    self.logger.info('Processing model output', length=len(model_output))
    self.metrics.increment('model_outputs_processed')
    
    try:
        # 解析模型输出
        parsed_result = self.parser.parse(model_output)
        
        # 记录解析结果
        self.logger.debug('Parsed model output', result=parsed_result)
        self.metrics.increment('model_outputs_parsed')
        
        # 处理工具调用请求
        if parsed_result['tool_calls']:
            return await self._handle_tool_calls(parsed_result['tool_calls'])
        else:
            # 直接返回结果给模型
            return model_output
    except Exception as e:
        self.logger.error('Error processing model output', error=str(e))
        self.metrics.increment('model_output_processing_errors')
        raise

async def _handle_tool_calls(self, tool_calls):
    """处理工具调用请求"""
    self.logger.info('Handling tool calls', count=len(tool_calls))
    self.metrics.increment('tool_calls_received', count=len(tool_calls))
    
    try:
        # 执行工具调用
        results = await self.executor.execute_tools(tool_calls)
        
        # 记录执行结果
        self.logger.debug('Tool execution results', results=results)
        self.metrics.increment('tool_calls_executed', count=len(results))
        
        # 更新上下文
        for tool_call, result in zip(tool_calls, results):
            self.context_manager.add_to_context({
                'type': 'tool_call',
                'tool_name': tool_call['tool_name'],
                'params': tool_call['params'],
                'result': result
            })
        
        # 格式化结果，返回给模型
        formatted_results = self._format_results(results)
        return formatted_results
    except Exception as e:
        self.logger.error('Error executing tool calls', error=str(e))
        self.metrics.increment('tool_call_execution_errors')
        raise

#!/usr/bin/env python3
"""
MCP Client 入门示例
"""

import asyncio
import uuid
import configparser
from mcp_client import MCClient

async def main():
    # 加载配置
    config = {
        'parser': {
            'supported_formats': ['json', 'yaml'],
            'default_format': 'json'
        },
        'executor': {
            'server_url': 'http://localhost:8080',
            'websocket_url': 'ws://localhost:8081',
            'use_websocket': False,
            'timeout': 30,
            'max_retries': 3,
            'parallel_execution': True
        },
        'context': {
            'max_context_length': 10000
        },
        'logging': {
            'level': 'INFO',
            'format': 'json'
        },
        'metrics': {
            'enabled': True
        }
    }
    
    # 创建 MCP Client 实例
    client = MCClient(config)
    
    # 启动 Client
    await client.start()
    
    try:
        # 模拟模型输出（包含工具调用请求）
        model_output = '''
        {
            "tool_call": {
                "name": "weather",
                "params": {
                    "location": "北京",
                    "date": "2026-01-01"
                }
            }
        }
        '''
        
        # 处理模型输出
        result = await client.process_model_output(model_output)
        print(f"处理结果: {result}")
        
        # 模拟多工具调用
        multi_tool_output = '''
        {
            "tool_calls": [
                {
                    "name": "weather",
                    "params": {
                        "location": "上海",
                        "date": "2026-01-01"
                    }
                },
                {
                    "name": "weather",
                    "params": {
                        "location": "广州",
                        "date": "2026-01-01"
                    }
                }
            ]
        }
        '''
        
        # 处理多工具调用
        multi_result = await client.process_model_output(multi_tool_output)
        print(f"多工具处理结果: {multi_result}")
        
        # 获取上下文摘要
        context_summary = client.context_manager.get_context_summary()
        print(f"上下文摘要: {context_summary}")
        
    finally:
        # 停止 Client
        await client.stop()

if __name__ == "__main__":
    asyncio.run(main())

class MCConfigManager:
    def __init__(self, config=None):
        self.config = config or {}
        self._load_from_file()
        self._load_from_env()
    
    def _load_from_file(self):
        """从配置文件加载配置"""
        import os
        import configparser
        
        config_file = os.environ.get('MCP_CONFIG_FILE', 'mcp_client.ini')
        if os.path.exists(config_file):
            parser = configparser.ConfigParser()
            parser.read(config_file)
            
            # 解析配置文件内容
            if 'parser' in parser:
                self.config.setdefault('parser', {})
                self.config['parser'].update(dict(parser['parser']))
            
            if 'executor' in parser:
                self.config.setdefault('executor', {})
                self.config['executor'].update(dict(parser['executor']))
            
            if 'context' in parser:
                self.config.setdefault('context', {})
                self.config['context'].update(dict(parser['context']))
    
    def _load_from_env(self):
        """从环境变量加载配置"""
        import os
        
        # 解析环境变量，支持嵌套配置
        for key, value in os.environ.items():
            if key.startswith('MCP_'):
                # 转换为嵌套配置
                parts = key[4:].lower().split('__')
                config = self.config
                
                for part in parts[:-1]:
                    config = config.setdefault(part, {})
                
                config[parts[-1]] = self._convert_value(value)
    
    def _convert_value(self, value):
        """转换环境变量值为合适的类型"""
        if value.lower() == 'true':
            return True
        elif value.lower() == 'false':
            return False
        elif value.isdigit():
            return int(value)
        elif '.' in value and all(part.isdigit() for part in value.split('.')):
            return float(value)
        return value
    
    def get(self, key, default=None):
        """获取配置值"""
        keys = key.split('.')
        config = self.config
        
        for k in keys:
            if k not in config:
                return default
            config = config[k]
        
        return config
    
    def set(self, key, value):
        """设置配置值"""
        keys = key.split('.')
        config = self.config
        
        for k in keys[:-1]:
            config = config.setdefault(k, {})
        
        config[keys[-1]] = value

class MCErrorHandler:
    def __init__(self, config=None):
        self.config = config or {}
        self.error_types = {
            'http_error': self._handle_http_error,
            'websocket_error': self._handle_websocket_error,
            'parse_error': self._handle_parse_error,
            'execution_error': self._handle_execution_error
        }
    
    def handle_error(self, error, error_type):
        """处理错误"""
        handler = self.error_types.get(error_type, self._handle_generic_error)
        return handler(error)
    
    def _handle_http_error(self, error):
        """处理 HTTP 错误"""
        import httpx
        
        if isinstance(error, httpx.HTTPStatusError):
            if error.response.status_code == 404:
                return {'error': 'Tool not found', 'status': 'error'}
            elif error.response.status_code == 403:
                return {'error': 'Access denied', 'status': 'error'}
            elif error.response.status_code == 500:
                return {'error': 'Internal server error', 'status': 'error'}
        return {'error': str(error), 'status': 'error'}
    
    def _handle_websocket_error(self, error):
        """处理 WebSocket 错误"""
        import websockets
        
        if isinstance(error, websockets.exceptions.ConnectionClosed):
            return {'error': 'WebSocket connection closed', 'status': 'error'}
        return {'error': str(error), 'status': 'error'}
    
    def _handle_parse_error(self, error):
        """处理解析错误"""
        return {
            'error': f'Failed to parse model output: {str(error)}',
            'status': 'error',
            'suggestion': 'Please check the model output format'
        }
    
    def _handle_execution_error(self, error):
        """处理执行错误"""
        return {'error': f'Tool execution failed: {str(error)}', 'status': 'error'}
    
    def _handle_generic_error(self, error):
        """处理通用错误"""
        return {'error': str(error), 'status': 'error'}

class MCLogger:
    def __init__(self, config=None):
        self.config = config or {}
        self.level = self.config.get('level', 'INFO')
        self.format = self.config.get('format', 'json')
        self._setup_logger()
    
    def _setup_logger(self):
        """设置日志配置"""
        import logging
        import structlog
        
        # 配置 structlog
        structlog.configure(
            processors=[
                structlog.contextvars.merge_contextvars,
                structlog.processors.add_log_level,
                structlog.processors.TimeStamper(fmt="iso"),
                structlog.processors.StackInfoRenderer(),
                structlog.dev.set_exc_info,
                structlog.processors.JSONRenderer() if self.format == 'json' else structlog.dev.ConsoleRenderer(),
            ],
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
            cache_logger_on_first_use=True,
        )
        
        # 配置标准日志
        root_logger = logging.getLogger()
        root_logger.setLevel(getattr(logging, self.level.upper()))
    
    def debug(self, message, **kwargs):
        """记录调试日志"""
        logger = structlog.get_logger()
        logger.debug(message, **kwargs)
    
    def info(self, message, **kwargs):
        """记录信息日志"""
        logger = structlog.get_logger()
        logger.info(message, **kwargs)
    
    def warning(self, message, **kwargs):
        """记录警告日志"""
        logger = structlog.get_logger()
        logger.warning(message, **kwargs)
    
    def error(self, message, **kwargs):
        """记录错误日志"""
        logger = structlog.get_logger()
        logger.error(message, **kwargs)
    
    def critical(self, message, **kwargs):
        """记录严重错误日志"""
        logger = structlog.get_logger()
        logger.critical(message, **kwargs)

class MCMetrics:
    def __init__(self, config=None):
        self.config = config or {}
        self.enabled = self.config.get('enabled', True)
        self.metrics = {}
        self._setup_metrics()
    
    def _setup_metrics(self):
        """设置指标收集器"""
        if not self.enabled:
            return
        
        try:
            from prometheus_client import Counter, Gauge, Histogram
            
            # 定义指标
            self.client_started = Counter('mcp_client_started_total', 'Number of times MCP Client started')
            self.client_stopped = Counter('mcp_client_stopped_total', 'Number of times MCP Client stopped')
            self.model_outputs_processed = Counter('mcp_model_outputs_processed_total', 'Number of model outputs processed')
            self.model_outputs_parsed = Counter('mcp_model_outputs_parsed_total', 'Number of model outputs successfully parsed')
            self.model_output_processing_errors = Counter('mcp_model_output_processing_errors_total', 'Number of model output processing errors')
            self.tool_calls_received = Counter('mcp_tool_calls_received_total', 'Number of tool calls received')
            self.tool_calls_executed = Counter('mcp_tool_calls_executed_total', 'Number of tool calls executed successfully')
            self.tool_call_execution_errors = Counter('mcp_tool_call_execution_errors_total', 'Number of tool call execution errors')
            self.context_length = Gauge('mcp_context_length', 'Current context length in characters')
            self.tool_execution_time = Histogram('mcp_tool_execution_time_seconds', 'Tool execution time in seconds')
        except ImportError:
            self.enabled = False
    
    def increment(self, metric_name, count=1):
        """增加计数器指标"""
        if not self.enabled:
            return
        
        metric = getattr(self, metric_name, None)
        if metric:
            metric.inc(count)
        else:
            # 简单的内存指标
            self.metrics.setdefault(metric_name, 0)
            self.metrics[metric_name] += count
    
    def gauge(self, metric_name, value):
        """设置 gauge 指标"""
        if not self.enabled:
            return
        
        metric = getattr(self, metric_name, None)
        if metric:
            metric.set(value)
        else:
            self.metrics[metric_name] = value
    
    def histogram(self, metric_name, value):
        """记录 histogram 指标"""
        if not self.enabled:
            return
        
        metric = getattr(self, metric_name, None)
        if metric:
            metric.observe(value)
        else:
            self.metrics.setdefault(metric_name, []).append(value)
    
    def get_metrics(self):
        """获取所有指标"""
        return self.metrics

# mcp_client.ini 配置示例

[parser]
supported_formats = json,yaml
default_format = json

[executor]
server_url = http://localhost:8080
websocket_url = ws://localhost:8081
use_websocket = false
timeout = 30
max_retries = 3
parallel_execution = true

[context]
max_context_length = 10000

[logging]
level = INFO
format = json

[metrics]
enabled = true

# 环境变量配置示例
export MCP_PARSER__SUPPORTED_FORMATS=json,yaml
export MCP_PARSER__DEFAULT_FORMAT=json
export MCP_EXECUTOR__SERVER_URL=http://localhost:8080
export MCP_EXECUTOR__USE_WEBSOCKET=false
export MCP_EXECUTOR__TIMEOUT=30
export MCP_EXECUTOR__MAX_RETRIES=3
export MCP_CONTEXT__MAX_CONTEXT_LENGTH=10000
export MCP_LOGGING__LEVEL=INFO
export MCP_METRICS__ENABLED=true