Kubernetes Tools MCP Server

Instructions

Implementation Reference

• src/k8stools/k8s_tools.py:450-500 (handler)

  Core handler function that fetches and formats Kubernetes events for the given pod using the Kubernetes Python client API.
  def get_pod_events(pod_name: str, namespace: str = "default") -> list[EventSummary]: """ Get events for a specific Kubernetes pod. This is equivalent to the kubectl command: `kubectl get events -n NAMESPACE --field-selector involvedObject.name=POD_NAME,involvedObject.kind=Pod` Parameters ---------- pod_name : str Name of the pod to retrieve events for. namespace : str, optional Namespace of the pod (default is "default"). Returns ------- list of EventSummary List of events associated with the specified pod. Each EventSummary has the following fields: last_seen : Optional[datetime.datetime] Timestamp of the last occurrence of the event (if available). type : str Type of the event. reason : str Reason for the event. object : str The object this event applies to. message : str Message describing the event. Raises ------ K8sConfigError If unable to initialize the K8S API. K8sApiError If the API call to list events fails. """ global K8S if K8S is None: K8S = _get_api_client() logging.info(f"get_pod_events(pod_name={pod_name}, namespace={namespace})") field_selector = f"involvedObject.name={pod_name}" events = K8S.list_namespaced_event(namespace, field_selector=field_selector) now = datetime.datetime.now(datetime.timezone.utc) return [ EventSummary( last_seen=(now - event.last_timestamp) if event.last_timestamp else None, type=event.type, reason=event.reason, object=getattr(event.involved_object, 'name', pod_name), message=event.message, ) for event in events.items ]
• src/k8stools/k8s_tools.py:441-447 (schema)

  Pydantic BaseModel defining the output schema for each event returned by get_pod_events.
  class EventSummary(BaseModel): """This is the representation of a Kubernetes Event""" last_seen: Optional[datetime.timedelta] # Time since event occurred type: str reason: str object: str message: str
• src/k8stools/k8s_tools.py:1101-1111 (registration)

  Registration of get_pod_events in the TOOLS list, which is imported and used 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-55 (registration)

  MCP server registration where TOOLS (including get_pod_events) are wrapped into Tool objects and passed to 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] mcp = FastMCP( name="k8stools-"+args.transport, tools=wrapped_tools,