get_container_logs
Retrieve logs from a specified container on a remote server by providing hostname, username, container details, and optional parameters like log tail or time range for troubleshooting and monitoring.
Instructions
获取指定容器的日志
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| container | No | ||
| hostname | Yes | ||
| password | No | ||
| port | No | ||
| since | No | ||
| tail | No | ||
| timeout | No | ||
| username | Yes |
Implementation Reference
- Primary handler implementation of the 'get_container_logs' MCP tool. Connects via SSH to the remote server, checks for Docker, executes 'docker logs' command with optional tail and since parameters, and returns structured log data.@handle_exceptions def get_container_logs( hostname: str, username: str, password: str = "", port: int = 22, container: str = "", # 容器ID或名称 tail: int = 100, # 获取最后多少行日志 since: str = "", # 从什么时间开始的日志，例如 "2023-01-01T00:00:00" timeout: int = 30 ) -> dict: """获取指定容器的日志""" result = InspectionResult() if not container: result.status = "error" result.error = "必须指定容器ID或名称" return result.dict() try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 检查Docker是否安装 stdin, stdout, stderr = ssh.exec_command("command -v docker") if not stdout.read().strip(): result.status = "error" result.error = "Docker未安装在目标服务器上" return result.dict() # 构建命令 cmd = f"docker logs --tail {tail}" if since: cmd += f" --since '{since}'" cmd += f" {container}" # 执行命令 stdin, stdout, stderr = ssh.exec_command(cmd) log_output = stdout.read().decode('utf-8') error_output = stderr.read().decode('utf-8') if error_output: result.status = "error" result.error = f"获取容器日志失败: {error_output}" return result.dict() # 设置结果 result.status = "success" result.data = {"logs": log_output.strip().split("\n")} result.raw_outputs = {"container_logs": log_output} log_lines = len(log_output.strip().split("\n")) if log_output.strip() else 0 result.summary = f"获取到容器 {container} 的 {log_lines} 行日志" except Exception as e: result.status = "error" result.error = f"获取容器日志失败: {str(e)}" return result.dict()
- SSE variant handler implementation of the 'get_container_logs' MCP tool, similar to the primary one but with minor differences in error handling and timeout usage.@handle_exceptions def get_container_logs( hostname: str, username: str, password: str = "", port: int = 22, container: str = "", # 容器ID或名称 tail: int = 100, # 获取最后多少行日志 since: str = "", # 从什么时间开始，如 "2021-01-01T00:00:00" timeout: int = 30 ) -> dict: """获取Docker容器的日志""" result = InspectionResult() if not container: result.status = "error" result.error = "必须指定容器ID或名称" return result.dict() try: with SSHManager(hostname, username, password, port, timeout) as ssh: # 检查Docker是否安装 stdin, stdout, stderr = ssh.exec_command("command -v docker", timeout=timeout) if not stdout.read().strip(): result.status = "error" result.error = "Docker未安装在目标服务器上" return result.dict() # 构建命令 cmd = f"docker logs --tail {tail}" if since: cmd += f" --since '{since}'" cmd += f" {container}" # 执行命令 stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout) log_output = stdout.read().decode('utf-8') error_output = stderr.read().decode('utf-8') if error_output: result.status = "error" result.error = f"获取容器日志失败: {error_output}" return result.dict() # 设置结果 result.status = "success" result.data = {"logs": log_output.strip().split("\n")} result.raw_outputs = {"container_logs": log_output} log_lines = len(log_output.strip().split("\n")) if log_output.strip() else 0 result.summary = f"获取到容器 {container} 的 {log_lines} 行日志" except Exception as e: result.status = "error" result.error = f"获取容器日志失败: {str(e)}" return result.dict()
- server_monitor/main.py:41-72 (registration)Registration of the 'get_container_logs' tool in the MCP server using FastMCP, mapping the function to the tool name in tools_dict and registering via decorator.# 注册所有工具函数 tools_dict = { 'get_memory_info': get_memory_info, 'remote_server_inspection': remote_server_inspection, 'get_system_load': get_system_load, 'monitor_processes': monitor_processes, 'check_service_status': check_service_status, 'get_os_details': get_os_details, 'check_ssh_risk_logins': check_ssh_risk_logins, 'check_firewall_config': check_firewall_config, 'security_vulnerability_scan': security_vulnerability_scan, 'backup_critical_files': backup_critical_files, 'inspect_network': inspect_network, 'analyze_logs': analyze_logs, 'list_docker_containers': list_docker_containers, 'list_docker_images': list_docker_images, 'list_docker_volumes': list_docker_volumes, 'get_container_logs': get_container_logs, 'monitor_container_stats': monitor_container_stats, 'check_docker_health': check_docker_health } # 使用装饰器动态注册所有工具 for name, func in tools_dict.items(): mcp.tool()(func) # 特殊处理list_available_tools，因为它需要mcp实例 @mcp.tool() def _list_available_tools(): return list_available_tools(mcp) return mcp
- server_monitor_sse/server.py:262-278 (registration)Dispatch registration in SSE server handler: calls get_container_logs function when tool name matches, with argument validation and parameter mapping.elif name == "get_container_logs": required_args = ["hostname", "username", "container"] for arg in required_args: if arg not in arguments: raise ValueError(f"Missing required argument '{arg}'") result = get_container_logs( hostname=arguments["hostname"], username=arguments["username"], password=arguments.get("password", ""), port=arguments.get("port", 22), container=arguments["container"], tail=arguments.get("tail", 100), since=arguments.get("since", ""), timeout=arguments.get("timeout", 30) )
- Schema definition includes ServerTools enum with CONTAINER_LOGS = 'get_container_logs' constant for tool identification and type safety.# ====================== class ServerTools(str, Enum): """服务器工具枚举""" MEMORY_INFO = "get_memory_info" REMOTE_INSPECTION = "remote_server_inspection" SSH_RISK_CHECK = "check_ssh_risk_logins" FIREWALL_CHECK = "check_firewall_config" OS_DETAILS = "get_os_details" SYSTEM_LOAD = "get_system_load" # 获取系统负载 LIST_TOOLS = "list_available_tools" # 列出可用工具 PROCESS_MONITOR = "monitor_processes" # 进程监控 SERVICE_STATUS = "check_service_status" # 服务状态检查 NETWORK_INSPECTION = "inspect_network" # 网络检查 LOG_ANALYZER = "analyze_logs" # 日志分析 FILE_BACKUP = "backup_critical_files" # 关键文件备份 SECURITY_SCAN = "security_vulnerability_scan" # 安全漏洞扫描 # 新增容器相关工具 DOCKER_CONTAINERS = "list_docker_containers" # 列出Docker容器 DOCKER_IMAGES = "list_docker_images" # 列出Docker镜像 DOCKER_VOLUMES = "list_docker_volumes" # 列出Docker卷 CONTAINER_LOGS = "get_container_logs" # 获取容器日志 CONTAINER_STATS = "monitor_container_stats" # 监控容器状态 DOCKER_HEALTHCHECK = "check_docker_health" # 检查Docker服务健康状态