mirrorneuron-python-sdk 1.1.2

MirrorNeuron Python SDK

MirrorNeuron Python SDK

Python client and workflow-bundle helpers for MirrorNeuron.

The SDK provides the gRPC client used by the CLI and API, plus decorators and bundle generation utilities for Python-defined workflows.

Features

• gRPC Client wrapper for job and runtime operations.
• Decorators for defining workflow and agent functions in Python.
• Bundle generation from decorated workflow classes.
• Runner, retry, and backpressure configuration helpers.
• Environment-based client configuration for local tools and services.

Tech Stack

Area	Tooling
Runtime	Python 3.11+
Transport	gRPC
Protocol	Protobuf
Packaging	setuptools with setuptools-scm

Prerequisites

• Python 3.11 or newer.
• A running MirrorNeuron core for client calls.
• Generated protocol modules included with the package.

Installation

The released-package installer installs this package automatically.

Standalone install:

pip install mirrorneuron-python-sdk

Developer install:

python3 -m venv .venv
source .venv/bin/activate
pip install -e .

Configuration

Constructor arguments take precedence over environment variables.

Variable	Default	Description
MN_GRPC_TARGET	localhost:50051	Core gRPC target when Client(target=...) is omitted.
MN_CORE_GRPC_TARGET	unset	Fallback core gRPC target.
MN_GRPC_TIMEOUT_SECONDS	10	Per-RPC timeout. Use 0 or none to disable.
MN_GRPC_AUTH_TOKEN	unset	Optional bearer metadata.
MN_SDK_LOG_PATH	~/.mn/logs/sdk.log	SDK log file path.
MN_LOG_LEVEL	package default	Log level used by shared logging setup.
MN_LOG_MAX_BYTES	package default	Rotating log size limit.
MN_LOG_BACKUP_COUNT	package default	Rotating log backup count.

Client Usage

from mn_sdk import Client

client = Client(target="localhost:50051")
job_id = client.submit_job(manifest_json="...", payloads={})

print(client.get_job(job_id))

Common client methods:

Method	Description
submit_job(manifest_json, payloads)	Submit a job from a manifest string and payload map.
get_job(job_id)	Fetch one job.
list_jobs(limit=0, include_terminal=True)	List jobs.
cancel_job(job_id)	Cancel a job.
pause_job(job_id)	Pause a job.
resume_job(job_id)	Resume a job.
clear_jobs()	Clear job history.
get_system_summary()	Fetch runtime summary.
remove_node(node_name)	Remove a node from the runtime view.
stream_events(job_id)	Stream job events.

Python Workflow Example

from pathlib import Path

from mn_sdk import RunnerConfig, agent, build_bundle, workflow


@agent.defn(type="map", runner=RunnerConfig.host_local())
def normalize_text(value: str) -> str:
    return value.strip().lower()


@workflow.defn(name="TextWorkflow")
class TextWorkflow:
    @workflow.run
    def run(self):
        normalize_text("  Example  ")


bundle_dir = build_bundle(TextWorkflow, Path("/tmp/text-workflow"))
print(bundle_dir)

The generated bundle can be submitted with the CLI:

mn run /tmp/text-workflow

Public API

The package exports:

• Client
• agent
• workflow
• build_bundle
• BundleCompileError
• RunnerConfig
• RetryPolicy
• BackpressurePolicy

Testing

python3 -m pytest -q

Deployment

Applications can depend on mirrorneuron-python-sdk directly when they need to submit jobs or generate bundles programmatically. Runtime deployment still requires the MirrorNeuron core and its backing services.

Troubleshooting

Symptom	Check
gRPC connection fails	Confirm the core is running and MN_GRPC_TARGET points to it.
Calls hang longer than expected	Set MN_GRPC_TIMEOUT_SECONDS.
Authenticated gateway rejects calls	Set MN_GRPC_AUTH_TOKEN.
Bundle generation fails	Check that the workflow class is decorated and that referenced agent calls are statically discoverable.

Contributing

Keep SDK changes backward-compatible with the CLI and API. Add focused tests for client calls, bundle generation, and configuration behavior.

License

MIT.