Splunk Enterprise SDK for Python

The Splunk SDK for Python is organized into several functional modules that allow developers to interact with Splunk Enterprise. The core modules include:

• splunklib.client: The primary interface for interacting with Splunk services, managing entities (like Users, Indexes, and Jobs), and collections.
• splunklib.binding: Low-level connectivity and HTTP handling, including custom HTTP handlers and authentication.
• splunklib.results: Tools for reading and parsing search results, such as JSONResultsReader.
• splunklib.data: Utilities for loading and working with Record objects.
• splunklib.modularinput: Framework for creating Modular Inputs to bring external data into Splunk.
• splunklib.searchcommands: Framework for creating custom Search Commands (Generating, Reporting, Streaming, and Eventing commands).
• splunklib.ai: An agentic framework for integrating AI models (Anthropic, OpenAI, Google) and managing agentic workflows, messages, and middleware.

The splunklib.modularinput package provides a Pythonic interface for developing Splunk Modular Inputs. It abstracts the complexities of the Modular Input framework, allowing you to define input schemes, handle arguments, and write events to Splunk using structured classes.

Key components include:

• Script: The core class used to define the lifecycle and execution of your modular input script.
• InputDefinition: Used to define the configuration interface for your input.
• Scheme: Defines the expected configuration parameters (arguments) for the input.
• Argument: Represents an individual configuration parameter within a Scheme.
• Event: Represents a single data record to be sent to Splunk.
• EventWriter: The interface used to stream Event objects into Splunk.
• ValidationDefinition: Used to define how input configurations should be validated.

To implement a modular input, you typically subclass Script and implement its required methods to handle configuration and data generation.

The SDK automatically selects one of two strategies for generating structured output based on the model's capabilities:

1. Provider strategy: Used when the model natively supports structured output. The validation error is fed back to the model to regenerate the output within the same agentic loop iteration.
2. Tool strategy: Used as a fallback. The LLM passes the structured output into a tool call. The tool schema corresponds to the output_schema Pydantic model. If validation fails, the error is returned as a tool response, prompting the model to retry the tool call in the same iteration.

Error Handling & Retries:

• If the model produces invalid output (wrong type, missing fields, or failing Pydantic validators), the SDK automatically retries until a valid output is generated or the maximum retry limit is reached.
• A StructuredOutputGenerationException is raised internally during failed attempts. You can intercept this using custom model_middleware to log or override retry behavior.

By default, agent.invoke is stateless. To enable the agent to persist and recall message history across multiple calls, you must provide a conversation_store.

InMemoryStore is a built-in implementation that keeps conversation history in the process memory. This is useful for maintaining context in a single session or across multiple calls within the same execution context.

from splunklib.ai import Agent, OpenAIModel
from splunklib.ai.conversation_store import InMemoryStore
from splunklib.ai.messages import HumanMessage
from splunklib.client import connect

model = OpenAIModel(...) 
service = connect(...) 

async with Agent(
    model=model,
    service=service,
    system_prompt="",
    conversation_store=InMemoryStore(),
) as agent:
    await agent.invoke([HumanMessage(content="Hi, my name is Chris.")])
    result = await agent.invoke([HumanMessage(content="What is my name?")])
    print(result.final_message.content)  # Chris

You can enable multi-turn conversations between a supervisor agent and a subagent by providing the subagent with its own conversation_store. This allows the supervisor to resume prior conversations with that specific subagent across multiple agent loop invocations.

Use InMemoryStore from splunklib.ai.conversation_store for a simple in-memory implementation.

from splunklib.ai import Agent, OpenAIModel
from splunklib.ai.conversation_store import InMemoryStore
from splunklib.client import connect

model = OpenAIModel(...)
service = connect(...)

async with (
    Agent(
        model=model,
        service=service,
        system_prompt="You are a log analysis expert.",
        name="log_analyzer_agent",
        description="Analyzes logs.",
        conversation_store=InMemoryStore(),
    ) as log_analyzer_agent,
):
    async with Agent(
        model=model,
        service=service,
        system_prompt="You are a supervisor.",
        agents=[log_analyzer_agent],
    ) as agent:
        await agent.invoke(...)

The splunklib.searchcommands module provides a framework for building custom Splunk search commands. To create a command, you should inherit from one of the specialized base classes depending on how your command handles data flow:

• StreamingCommand: For commands that process events one by one as they pass through the pipeline (e.g., filtering or modifying individual events). Implement the stream method.
• ReportingCommand: For commands that aggregate data (e.g., calculating statistics). These support a MapReduce-style pattern using map and reduce methods.
• GeneratingCommand: For commands that create new events from scratch rather than consuming input events (e.g., | makeresults). Implement the generate method.
• EventingCommand: For commands that perform complex transformations on event sets.

Once your class is defined, use the dispatch function to execute the command, which handles the integration with Splunk's input/output streams.

from splunklib.searchcommands import StreamingCommand, dispatch

class MyStreamingCommand(StreamingCommand):
    def stream(self, events):
        for event in events:
            # Modify event logic here
            yield event

# To run the command
if __name__ == '__main__':
    dispatch(MyStreamingCommand)

To avoid 'context bloat' and optimize for cost and capability, you can break complex workflows into specialized Agent instances called subagents.

By passing a list of subagents to the agents parameter of a 'supervisor' agent, the supervisor can delegate tasks. The supervisor decides which subagent to call based on the subagent's name and description.

Key configuration for subagents:

• name: A unique identifier for the agent.
• description: A text explanation of what the agent does (used by the supervisor for routing).
• tool_settings: Defines which tools the agent can access via ToolSettings, LocalToolSettings, and ToolAllowlist.

from splunklib.ai import Agent, OpenAIModel
from splunklib.ai.messages import HumanMessage
from splunklib.ai.tool_settings import LocalToolSettings, ToolAllowlist, ToolSettings
from splunklib.client import connect

model = OpenAIModel(...) 
service = connect(...) 

# Define specialized subagents
async with (
    Agent(
        model=highly_specialized_model,
        service=service,
        system_prompt="You are a highly specialized debugging agent...",
        name="debugging_agent",
        description="Agent, that provided with logs will analyze and debug complex issues",
        tool_settings=ToolSettings(
            local=LocalToolSettings(allowlist=ToolAllowlist(tags=["debugging"])),
            remote=None,
        ),
    ) as debugging_agent,
    Agent(
        model=low_cost_model,
        service=service,
        system_prompt="You are a log analyzer agent...",
        name="log_analyzer_agent",
        description="Agent, that provided with a problem details will return logs...",
        tool_settings=ToolSettings(
            local=LocalToolSettings(allowlist=ToolAllowlist(tags=["spl"])),
            remote=None,
        ),
    ),
) as (debugging_agent, log_analyzer_agent):
    # The supervisor agent uses the subagents to perform tasks
    async with Agent(
        model=low_cost_model,
        service=service,
        system_prompt="You are a supervisor agent, use available subagents to perform requested operations.",
        agents=[debugging_agent, log_analyzer_agent],
    ) as agent:
        result = await agent.invoke(
            [
                HumanMessage(
                    content="Query the logs in the index 'main', and try to debug the root cause of this issue."
                )
            ]
        )