Python SDK
The Monad Python SDK provides a robust client interface for interacting with the Monad API.
Installation
To use the Monad Python SDK in your project, clone the Monad Python SDK, and run pip install in the root directory of the python SDK.
Configuration
Configuration Options
The Configuration class accepts the following parameters:
host: API host URLapi_key: Dictionary containing API key configurationdebug: Setting to true logs all HTTP request and response details.
Example Configuration
import monad
configuration = monad.Configuration(
host="app.monad.com/api",
api_key={
'ApiKeyAuth': 'YOUR_API_KEY'
},
debug=True
)
client = monad.ApiClient(configuration)
The API Key can be obtained from the Monad UI in the organizations page.
API Services
Monad provides access to several APIs to perform fundamental operations through the client created above, such as creating inputs through the Organization Inputs API, creating Outputs from the Organization Outputs API, transforms with the Organization Transforms API, creating Pipelines with the Pipelines API, and more.
Type-Safe Component System
The Monad Python SDK implements a comprehensive type system that ensures type safety across all components (inputs, outputs, and transforms). Each component type has its own dedicated configuration classes that enforce proper configuration.
Complete Pipeline Example
Below is a complete example that demonstrates creating a pipeline with a Semgrep input, a Splunk output, and a Timestamp transform using the Python SDK. The example shows how to create all components and connect them in a single pipeline.
import monad
from uuid import uuid4
from monad.models.semgrep_projects_secrets_config import SemgrepProjectsSecretsConfig
from monad.models.semgrep_projects_settings_config import SemgrepProjectsSettingsConfig
from monad.models.splunk_secrets_config import SplunkSecretsConfig
from monad.models.splunk_settings_config import SplunkSettingsConfig
def create_pipeline(org_id: str):
# Initialize the Monad client configuration
configuration = monad.Configuration(
host="app.monad.com/api",
api_key={
'ApiKeyAuth': 'YOUR_API_KEY'
},
debug=True ## Setting to true logs all HTTP request and response details.
)
# Create API client
with monad.ApiClient(configuration) as client:
try:
# Step 1: Create Semgrep input
input_name = f"semgrep-input-{str(uuid4())}"
# Configure Semgrep settings and secrets
semgrep_secrets = '{"api_key": "YOUR_SEMGREP_API_KEY"}'
semgrep_settings = '{"cron": "* * * * *"}'
semgrep_secrets_instance = SemgrepProjectsSecretsConfig.from_json(semgrep_secrets)
semgrep_settings_instance = SemgrepProjectsSettingsConfig.from_json(semgrep_settings)
input_config = monad.RoutesV2InputConfig(
secrets=monad.RoutesV2InputConfigSecrets(
semgrep_secrets_instance
),
settings=monad.RoutesV2InputConfigSettings(
semgrep_settings_instance
)
)
input_request = monad.RoutesV2CreateInputRequest(
name=input_name,
type="semgrep-projects",
config=input_config
)
inputs_api = monad.OrganizationInputsApi(client)
input_response = inputs_api.v2_organization_id_inputs_post(
org_id,
input_request
)
# Step 2: Create Splunk output
output_name = f"splunk-output-{str(uuid4())}"
# Configure Splunk settings and secrets
splunk_secrets = '{"token": "YOUR_SPLUNK_TOKEN"}'
splunk_settings = '{"url": "YOUR_SPLUNK_URL", "port": "443", "allow_insecure": false}'
splunk_secrets_instance = SplunkSecretsConfig.from_json(splunk_secrets)
splunk_settings_instance = SplunkSettingsConfig.from_json(splunk_settings)
output_config = monad.RoutesV2OutputConfig(
secrets=monad.RoutesV2OutputConfigSecrets(
splunk_secrets_instance
),
settings=monad.RoutesV2OutputConfigSettings(
splunk_settings_instance
)
)
output_request = monad.RoutesV2CreateOutputRequest(
name=output_name,
output_type="splunk",
config=output_config
)
outputs_api = monad.OrganizationOutputsApi(client)
output_response = outputs_api.v2_organization_id_outputs_post(
org_id,
output_request
)
# Step 3: Create timestamp transform
transform_name = f"timestamp-transform-{str(uuid4())}"
transform_operation = monad.ModelsTransformOperation(
operation="timestamp",
arguments={
"key": "timestamp",
"format": "RFC3339"
}
)
transform_config = monad.ModelsTransformConfig(
operations=[transform_operation]
)
transform_request = monad.RoutesCreateTransformRequest(
name=transform_name,
config=transform_config
)
transform_api = monad.OrganizationTransformsApi(client)
transform_response = transform_api.v1_organization_id_transforms_post(
org_id,
transform_request
)
# Step 4: Create the pipeline connecting all components
input_slug = "input-semgrep"
transform_slug = "transform-timestamp"
output_slug = "output-splunk"
# Defines the pipeline nodes (components)
input_node = monad.RoutesV2PipelineRequestNode(
component_id=input_response.id,
component_type="input",
enabled=True,
slug=input_slug
)
transform_node = monad.RoutesV2PipelineRequestNode(
component_id=transform_response.id,
component_type="transform",
enabled=True,
slug=transform_slug
)
output_node = monad.RoutesV2PipelineRequestNode(
component_id=output_response.id,
component_type="output",
enabled=True,
slug=output_slug
)
# Configure edge conditions
edge_conditions = monad.ModelsPipelineEdgeConditions(
operator="always"
)
# Defines the connections between nodes
edge1 = monad.RoutesV2PipelineRequestEdge(
from_node_instance_id=input_slug,
to_node_instance_id=transform_slug,
name=f"edge-{str(uuid4())}",
description="Edge connecting Semgrep input to timestamp transform",
conditions=edge_conditions
)
edge2 = monad.RoutesV2PipelineRequestEdge(
from_node_instance_id=transform_slug,
to_node_instance_id=output_slug,
name=f"edge-{str(uuid4())}",
description="Edge connecting timestamp transform to Splunk output",
conditions=edge_conditions
)
# Create pipeline request
pipeline_request = monad.RoutesV2CreatePipelineRequest(
name=f"semgrep-to-splunk-pipeline-{str(uuid4())}",
description="Pipeline connecting Semgrep input to Splunk output with timestamp transform",
enabled=True,
nodes=[input_node, transform_node, output_node],
edges=[edge1, edge2]
)
# Create pipeline
pipeline_api = monad.PipelinesApi(client)
pipeline_response = pipeline_api.v2_organization_id_pipelines_post(
org_id,
pipeline_request
)
return pipeline_response
except monad.ApiException as e:
print(f"Error creating pipeline: {e}")
raise
if __name__ == "__main__":
org_id = "YOUR_ORG_ID"
pipeline = create_pipeline(org_id)
This example demonstrates:
- Creating and configuring all components in a single flow
- Using descriptive slugs for pipeline nodes
- Proper error handling with try/except blocks
- Clear component connections using edges
- Type-safe configuration for all components
- Context manager usage for proper resource cleanup
Support
For additional support contact support@monad.com