Kubernetes Tools MCP Server

Instructions

Implementation Reference

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

  Core handler function implementing the tool logic using Kubernetes client to fetch pod/container logs.
  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)

  Registration of the tool function in the TOOLS list for MCP server integration.
  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/mock_tools.py:309-323 (handler)

  Mock handler for testing purposes, mimicking the real tool behavior.
  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)

  Mock tools registration list including the mock get_logs_for_pod_and_container.
  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 ]