splunk-mcp

Instructions

Implementation Reference

• splunk_mcp.py:333-378 (handler)

  The main handler function decorated with @mcp.tool() that executes Splunk search queries, handles parameters, connects to Splunk, runs the job, and returns results. This serves as both the handler and registration point.
  @mcp.tool() async def search_splunk(search_query: str, earliest_time: str = "-24h", latest_time: str = "now", max_results: int = 100) -> List[Dict[str, Any]]: """ Execute a Splunk search query and return the results. Args: search_query: The search query to execute earliest_time: Start time for the search (default: 24 hours ago) latest_time: End time for the search (default: now) max_results: Maximum number of results to return (default: 100) Returns: List of search results """ if not search_query: raise ValueError("Search query cannot be empty") # Prepend 'search' if not starting with '|' or 'search' (case-insensitive) stripped_query = search_query.lstrip() if not (stripped_query.startswith('|') or stripped_query.lower().startswith('search')): search_query = f"search {search_query}" try: service = get_splunk_connection() logger.info(f"🔍 Executing search: {search_query}") # Create the search job kwargs_search = { "earliest_time": earliest_time, "latest_time": latest_time, "preview": False, "exec_mode": "blocking" } job = service.jobs.create(search_query, **kwargs_search) # Get the results result_stream = job.results(output_mode='json', count=max_results) results_data = json.loads(result_stream.read().decode('utf-8')) return results_data.get("results", []) except Exception as e: logger.error(f"❌ Search failed: {str(e)}") raise
• splunk_mcp.py:298-332 (helper)

  Helper function used by search_splunk to establish a connection to the Splunk service supporting both username/password and token authentication.
  def get_splunk_connection() -> splunklib.client.Service: """ Get a connection to the Splunk service. Supports both username/password and token-based authentication. If SPLUNK_TOKEN is set, it will be used for authentication and username/password will be ignored. Returns: splunklib.client.Service: Connected Splunk service """ try: if SPLUNK_TOKEN: logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} using token authentication") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL, token=f"Bearer {SPLUNK_TOKEN}" ) else: username = os.environ.get("SPLUNK_USERNAME", "admin") logger.debug(f"🔌 Connecting to Splunk at {SPLUNK_SCHEME}://{SPLUNK_HOST}:{SPLUNK_PORT} as {username}") service = splunklib.client.connect( host=SPLUNK_HOST, port=SPLUNK_PORT, username=username, password=SPLUNK_PASSWORD, scheme=SPLUNK_SCHEME, verify=VERIFY_SSL ) logger.debug(f"✅ Connected to Splunk successfully") return service except Exception as e: logger.error(f"❌ Failed to connect to Splunk: {str(e)}") raise
• splunk_mcp.py:47-52 (registration)

  Initialization of the FastMCP server instance where tools like search_splunk are registered via decorators.
  mcp = FastMCP( "splunk", description="A FastMCP-based tool for interacting with Splunk Enterprise/Cloud through natural language", version="0.3.0", host="0.0.0.0", # Listen on all interfaces port=FASTMCP_PORT