IronShell Python SDK

Python bindings for IronShell - AI-powered typed shell with workflow orchestration and cloud deployment.

Installation

# Core SDK
pip install ironshell

# With LangChain integration
pip install ironshell[langchain]

# With cloud deployment support
pip install ironshell[cloud]

# Everything
pip install ironshell[all]

Or from a checkout of the repository:

pip install ./integrations/python

Requires the ish binary on PATH. Install it with cargo install iron-shell, or take a prebuilt binary from the releases page.

Versioning: this SDK versions independently of the ish shell. The SDK in this repository is 1.5.1; PyPI carries 1.5.0, a manual upload, because CI has never published it. pip install ./integrations/python installs the current one.

Quick Start

from ironshell import IronShellRuntime, Agent

# Create runtime
runtime = IronShellRuntime()

# Evaluate IronShell code
result = runtime.eval('[1, 2, 3] | map(fn(x) => x * 2)')
print(result)  # [2, 4, 6]

# Create and run an agent
agent = runtime.create_agent(
    name="researcher",
    tools=["http_get", "search"]
)

result = await agent.run("Find the latest Python release version")
print(result)

Features

• Evaluation: Execute IronShell code from Python
• Pipelines: Process data with typed pipelines
• Agents: Create and orchestrate AI agents
• Swarms: Multi-agent coordination
• Workflows: MapReduce, Saga, Fan-Out patterns
• Metrics: Prometheus metrics, tracing, health checks
• Distributed: Service discovery, leader election, load balancing
• Cloud: Deploy as serverless functions (AWS, Azure, GCP, K8s)
• LangChain: Full LangChain tool integration

Workflow Orchestration

from ironshell.workflows import (
    MapReduceWorkflow,
    SagaWorkflow,
    PipelineWorkflow,
    CircuitBreaker,
)

# MapReduce for parallel processing
workflow = MapReduceWorkflow(
    name="word_count",
    map_fn=lambda text: len(text.split()),
    reduce_fn=lambda a, b: a + b,
)
result = await workflow.run(["hello world", "foo bar baz"])
print(result.result)  # 5

# Saga with compensation
saga = SagaWorkflow("order")
saga.add_saga_step("reserve", reserve_inventory, rollback_reservation)
saga.add_saga_step("charge", charge_payment, refund_payment)
saga.add_saga_step("ship", ship_order, cancel_shipment)
result = await saga.run(order_data)

# Circuit breaker for fault tolerance
breaker = CircuitBreaker(name="api", failure_threshold=5)
result = breaker.call(lambda: api_request())

Metrics & Observability

from ironshell.metrics import (
    MetricsCollector,
    Counter,
    Gauge,
    Histogram,
    Tracer,
)

# Create metrics
collector = MetricsCollector(namespace="myapp")
requests = collector.counter("requests_total")
latency = collector.histogram("request_latency_seconds")

# Track metrics
requests.inc()
latency.observe(0.125)

# Export to Prometheus
print(collector.to_prometheus())

# Distributed tracing
tracer = collector.tracer("my-service")
with tracer.start_span("handle_request") as span:
    span.set_attribute("user_id", "123")
    # ... process request

Distributed Agents

from ironshell.distributed import (
    ServiceRegistry,
    LeaderElection,
    LoadBalancer,
    DistributedSwarm,
)

# Service discovery
registry = ServiceRegistry()
registry.register("agent-nlp", "host1", 8080)
registry.register("agent-nlp", "host2", 8080)

# Load balancing
lb = LoadBalancer(registry, strategy="round_robin")
service = lb.select_service("agent-nlp")

# Leader election
election = LeaderElection("node-1", registry, "cluster")
await election.run_election()
if election.is_leader:
    print("I am the leader!")

# Distributed swarm
swarm = DistributedSwarm("my-swarm", registry)
swarm.add_local_agent(my_agent)
result = await swarm.dispatch(goal="analyze data", capability="nlp")

Cloud Deployment

Deploy agents as serverless functions:

from ironshell.cloud import (
    CloudProvider,
    FunctionConfig,
    DeploymentConfig,
    deploy_agent,
)

# Configure function
config = DeploymentConfig(
    provider=CloudProvider.AWS_LAMBDA,
    region="us-east-1",
    function_config=FunctionConfig(
        name="my-agent",
        memory_mb=512,
        timeout_seconds=60,
    ),
)

# Generate deployment files
agent_code = '''
def create_agent(runtime):
    return Agent(name="analyst", runtime=runtime)
'''

files = deploy_agent(config, agent_code, output_dir="./deploy")
# Creates: handler.py, template.yaml, samconfig.toml, requirements.txt

Supported platforms:

• AWS Lambda (SAM template)
• Azure Functions (Bicep)
• GCP Cloud Functions (Terraform)
• Kubernetes/Knative (manifests + Skaffold)

LangChain Integration

from ironshell.langchain import (
    get_all_ironshell_tools,
    IronShellWorkflowTool,
    IronShellMapReduceTool,
    IronShellMetricsTool,
)

# Get all tools for LangChain agent
tools = get_all_ironshell_tools()

# Use with LangChain
from langchain.agents import initialize_agent
agent = initialize_agent(tools, llm, agent="zero-shot-react-description")
agent.run("Process this data with MapReduce: [1,2,3,4,5]")

API Reference

IronShellRuntime

runtime = IronShellRuntime()

# Evaluate code
result = runtime.eval(code: str) -> Any

# Create agent
agent = runtime.create_agent(
    name: str,
    model: str = "openai:gpt-4o-mini",
    tools: List[str] = [],
    max_steps: int = 10
) -> Agent

# Run swarm (sent to the swarm builtin as {goal, tools, max_steps};
# tools is the union of the agents' tools; policy must be "round_robin",
# and agents must use the default model and max_steps, else ValueError)
result = await runtime.run_swarm(
    agents: List[Agent],
    goal: str,
    policy: str = "round_robin",
    max_iterations: int = 10
) -> SwarmResult   # result.result is the builtin's combined answer

Agent

agent = Agent(name="agent1")  # model must stay at the default openai:gpt-4o-mini

# Run agent (sent as agent({goal, tools, max_steps}); result.result is the
# final answer, and trace and steps_taken are empty)
result = await agent.run(goal: str) -> AgentResult

# Get trace
trace = agent.trace  # List of steps taken

A2UI Events

# Subscribe to events
def on_event(event: A2UIEvent):
    print(f"Event: {event.type}")

runtime.subscribe_a2ui(on_event)

# Event types
# - Notify: Notifications
# - Progress: Progress updates
# - Prompt: User prompts
# - AgentThinking: Agent reasoning

Development

# Clone repository
git clone https://github.com/nervosys/IronShell.git
cd IronShell/integrations/python

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Build package
python -m build

License

AGPL-3.0-or-later with commercial dual-license option — see LICENSE for details.