Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:806-853 (handler)

  The primary handler function implementing the tool logic. It uses the Kubernetes CoreV1Api client to read logs from the specified pod and optional container, with limits on lines and bytes to prevent memory issues.
  def get_logs_for_pod_and_container(pod_name:str, namespace:str = "default", container_name:Optional[str]=None) -> Optional[str]: """ Retrieves logs from a Kubernetes pod and container. Args: namespace (str): The namespace of the pod. pod_name (str): The name of the pod. container_name (str, optional): The name of the container within the pod. If None, defaults to the first container. Returns: str, optional: Log content if any found for this pod/container, or None otherwise Raises ------ K8sConfigError If unable to initialize the K8S API. K8sApiError If the API call to fetch logs fails or an unexpected error occurs. """ global K8S if K8S is None: K8S = _get_api_client() try: # read_namespaced_pod_log with reasonable limits to avoid memory issues resp = K8S.read_namespaced_pod_log( name=pod_name, namespace=namespace, container=container_name, # Pass container_name if specified follow=False, # Set to False to get all current logs _preload_content=True, # Important: This loads all content into memory timestamps=True, # Optional: Include timestamps tail_lines=1000, # Limit to last 1000 lines to avoid memory issues limit_bytes=1024*1024 # Limit to 1MB to avoid memory issues ) # The response is a single string containing all logs if resp: return resp else: return '' except client.ApiException as e: raise K8sApiError(f"Error fetching logs: {e}") from e except Exception as e: raise K8sApiError(f"An unexpected error occurred: {e}") from e
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  The TOOLS list in k8s_tools.py exports the get_logs_for_pod_and_container function along with other tools for use by the MCP server.
  TOOLS = [ get_namespaces, get_node_summaries, get_pod_summaries, get_pod_container_statuses, get_pod_events, get_pod_spec, get_logs_for_pod_and_container, get_deployment_summaries, get_service_summaries ]
• src/k8stools/mcp_server.py:46-52 (registration)

  The MCP server script imports the TOOLS list from k8s_tools (or mock_tools if --mock) and wraps each function into a Tool object using Tool.from_function for registration in the FastMCP server.
  if not args.mock: from .k8s_tools import TOOLS else: from .mock_tools import TOOLS logging.warning(f"Using mock versions of the tools") wrapped_tools = [get_tool_for_function(fn) for fn in TOOLS]
• src/k8stools/mock_tools.py:309-323 (handler)

  Mock handler function used when --mock flag is provided to the MCP server. It returns static log data without requiring a real Kubernetes cluster and copies the docstring from the real handler.
  def get_logs_for_pod_and_container(pod_name: str, namespace: str = "default", container_name: Optional[str] = None) -> Optional[str]: """Mock implementation that returns static log data for the specified pod and container""" # For the specific ad pod, return cached data if pod_name == "ad-647b4947cc-s5mpm" and namespace == "default": return _MOCK_DATA['ad_pod_logs'] # For other pods, return generic mock logs container_ref = container_name or pod_name.split('-')[0] return f"""2025-07-28T01:30:00.000000000Z Starting {container_ref} container 2025-07-28T01:30:01.000000000Z {container_ref} container started successfully 2025-07-28T01:30:02.000000000Z Processing requests... 2025-07-28T01:30:03.000000000Z Ready to serve traffic""" get_logs_for_pod_and_container.__doc__ = k8s_tools.get_logs_for_pod_and_container.__doc__
• src/k8stools/mock_tools.py:348-358 (registration)

  The TOOLS list in mock_tools.py, used by the MCP server when --mock is specified.
  TOOLS = [ get_namespaces, get_node_summaries, get_pod_summaries, get_pod_container_statuses, get_pod_events, get_pod_spec, get_logs_for_pod_and_container, get_deployment_summaries, get_service_summaries ]