aiel-sdk 1.7.8

AI Execution Layer SDK (contracts + registry + decorators) with curated facades.

AIEL SDK

AI Execution Layer SDK — Enterprise-grade contract-first decorators, registry system, and curated facade for AI orchestration.

---

Overview

AIEL SDK provides a stable contract surface for building AI-powered applications that execute on the AIEL Execution Plane. This lightweight SDK enables:

• Contract-first development via decorators and type-safe registry
• Curated facade modules (aiel.*) mirroring server runtime imports
• Clear error messages for missing optional dependencies
• Type safety and IDE support for local development

Note: Code execution is performed server-side by the Execution Plane using aiel-runtime. This SDK enables local development, type-checking, linting, and testing with identical import paths used in production.

---

Table of Contents

• Installation
• Quick Start
• Core Concepts
• Platform Architecture
• API Reference
• Runtime Execution
• Best Practices
• Troubleshooting

---

Installation

Installation Matrix

Choose the appropriate installation based on your requirements:

Use Case	Command	Description
Minimal (Recommended)	pip install aiel-sdk	Core contracts, decorators, and registry only
LangGraph Development	pip install "aiel-sdk[langgraph]"	Adds LangGraph facade support
LangChain Development	pip install "aiel-sdk[langchain]"	Adds LangChain Core facade support
LangSmith Integration	pip install "aiel-sdk[langsmith]"	Adds LangSmith facade support
AIEL CLI Context	pip install "aiel-sdk[aiel-cli]"	Adds CLI user context facade support
Full Development Suite	pip install "aiel-sdk[all]"	All curated integrations (best for experimentation)

Shell Note: When using zsh, always quote extras: pip install "aiel-sdk[all]"

Requirements

• Python 3.8 or higher
• pip 21.0 or higher

---

Quick Start

Basic Example

from aiel_sdk import tool, agent, flow, http, mcp_server
from aiel_core.pydantic import BaseModel, Field

# Define data models
class NormalizeEmailPayload(BaseModel):
    email: str = Field(..., description="Raw email address")
    name: str = Field(..., description="User's full name")

class NormalizeEmailOut(BaseModel):
    email: str = Field(..., description="Normalized email address")

# Define a tool
@tool("normalize_email")
def normalize_email(ctx, payload: dict) -> dict:
    """Normalize email addresses to lowercase."""
    data = NormalizeEmailPayload.model_validate(payload)
    email = (data.email or "").strip().lower()
    ctx.log("normalize_email", email=email)
    return NormalizeEmailOut(email=email).model_dump()

# Define an agent
@agent("collect_personal_data")
def collect_personal_data(ctx, state: dict) -> dict:
    """Collect and validate personal information."""
    email_info = normalize_email(ctx, {
        "email": state.get("email"),
        "name": state.get("name")
    })
    state = dict(state)
    state["email"] = email_info["email"]
    state["personal_validated"] = True
    return state

# Define a flow
@flow("driver_onboarding")
def driver_onboarding(ctx, input: dict) -> dict:
    """Main onboarding orchestration flow."""
    state = collect_personal_data(ctx, input)
    return {"status": "ok", "state": state}

# Define HTTP endpoint
@http.post("/driver/onboard")
def http_driver_onboard(ctx, body: dict) -> dict:
    """HTTP handler for driver onboarding."""
    return driver_onboarding(ctx, body)

# Expose via MCP
mcp_server("driver_support", tools=["normalize_email"])

Using Facade Imports

from aiel_core.langgraph.graph import StateGraph, MessagesState, START, END
from aiel_core.langchain.agents import create_agent

# Build a state graph
graph = StateGraph(MessagesState)
graph.add_node("process", lambda state: state)
graph.add_edge(START, "process")
graph.add_edge("process", END)

app = graph.compile()
result = app.invoke({"messages": []})

Backend Integrations (Jira, Slack, Postgres)

from aiel_sdk import IntegrationsClient

client = IntegrationsClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
)

connections = client.list("workspace_id", "project_id")
first = connections[0]

detail = client.get(first.workspace_id, first.project_id, first.connection_id)

# Example action (Slack)
client.invoke_action(
    first.workspace_id,
    first.project_id,
    first.connection_id,
    action="slack.post_message",
    payload={"params": {"channel": "#general", "text": "Hello from AIEL SDK"}},
)

---

Core Concepts

Contract-First Architecture

The SDK defines a stable contract surface that mirrors the server runtime environment. This approach ensures:

• Local Development Parity: Import paths work identically in local and production environments
• Type Safety: Full IDE support and type checking during development
• Clear Contracts: Explicit interfaces between your code and the Execution Plane

Registry System

Decorators automatically register exports into an in-memory registry:

• @tool(name) — Discrete actions callable by agents and flows
• @agent(name) — Orchestration steps that operate on state
• @flow(name) — Main orchestration entry points
• @http.get(path) / @http.post(path) — HTTP handler exports
• mcp_server(name, tools=[]) — MCP server exposure metadata

Facade Modules

The aiel.* namespace provides stable import paths for curated integrations:

Module	Import Path	Purpose
Pydantic	aiel_core.pydantic	Data validation and serialization
LangGraph	aiel_core.langgraph.graph	State graph orchestration
LangChain	aiel_core.langchain.core	Prompt templates and runnables
LangChain Messages	aiel_core.langchain.messages	AI/Human message types
LangSmith	aiel_core.langsmith.client	Observability and tracing
AIEL CLI	aiel_core.aiel_cli.context.user_context	CLI user context helpers

Missing integrations fail with actionable error messages indicating the required installation command.

---

Platform Architecture

AIEL is easiest to reason about as a four-layer system:

1. Control Plane

The Control Plane orchestrates publish, release, and promotion decisions.

• Accepts new versions and deployment intents
• Decides whether a version can move from candidate to production
• Reads evidence produced by the Engineering Layer and telemetry stored in the Data Plane
• Triggers canary rollout and rollback when thresholds are crossed

2. Data Plane

The Data Plane stores the artifacts and evidence required to operate the system safely.

• Project snapshots and manifests
• Eval baselines and regression reports
• Traces, metrics, and rollout state
• Policy outcomes and release metadata

3. Runtime

The Runtime is the execution environment for customer code.

• Loads the application entry point
• Resolves curated runtime dependencies
• Executes tools, agents, flows, and HTTP handlers
• Produces traces and runtime metrics

4. Engineering Layer

The Engineering Layer is what makes AI systems shippable. It is intentionally narrower than a general CI platform and focuses on AI release safety:

• Offline evals
• Regression tests
• Tracing
• Rollout gates
• Policy checks
• Canary and rollback thinking

This layer answers one question: should this version be promoted, and if it is promoted, should it stay live?

Release Flow: PR to Production

For AIEL-style systems, a concrete release flow looks like this:

1. A new version is proposed.
2. The Engineering Layer runs offline evals, regression checks, policy checks, and rollout gating.
3. The Control Plane reads those results and decides whether the version is eligible for promotion.
4. The Runtime executes the candidate in a controlled rollout.
5. The Data Plane records traces, metrics, policy outcomes, and rollout state.
6. The Control Plane either promotes fully, keeps the rollout limited, or rolls back.

This keeps responsibilities clear:

• The Control Plane decides
• The Data Plane remembers
• The Runtime runs
• The Engineering Layer judges ship-readiness and rollback-readiness

Current SDK Scope

This SDK already includes core Engineering Layer building blocks:

• evals
• policy
• regression
• rollout
• rollback
• tracing

These primitives are designed to support AI release engineering rather than a full generic build system. Standard software concerns such as linting, unit-test orchestration, and CI runner management can exist around AIEL, but they are not the primary responsibility of the Engineering Layer defined here.

---

API Reference

Decorators

@tool(name: str)

Register a discrete tool that can be invoked by agents and flows.

@tool("my_tool")
def my_tool(ctx, payload: dict) -> dict:
    """
    Args:
        ctx: Execution context with logging and utilities
        payload: Input data dictionary
    
    Returns:
        Output data dictionary
    """
    ctx.log("operation", key="value")
    return {"result": "success"}

@agent(name: str)

Register an orchestration agent that operates on state.

@agent("my_agent")
def my_agent(ctx, state: dict) -> dict:
    """
    Args:
        ctx: Execution context
        state: Current state dictionary
    
    Returns:
        Updated state dictionary
    """
    return {**state, "processed": True}

@flow(name: str)

Register a main orchestration entry point.

@flow("my_flow")
def my_flow(ctx, input: dict) -> dict:
    """
    Args:
        ctx: Execution context
        input: Initial input dictionary
    
    Returns:
        Final output dictionary
    """
    return {"status": "complete"}

@http.get(path: str) / @http.post(path: str)

Register HTTP endpoint handlers.

@http.get("/status")
def get_status(ctx, query: dict) -> dict:
    """Handle GET requests."""
    return {"status": "healthy"}

@http.post("/process")
def process_data(ctx, body: dict) -> dict:
    """Handle POST requests."""
    return {"processed": True}

flow_graph(name: str, builder_fn: Callable)

Register a LangGraph-based flow.

def build_graph():
    graph = StateGraph(dict)
    graph.add_node("start", lambda s: s)
    graph.add_edge(START, "start")
    graph.add_edge("start", END)
    return graph.compile()

flow_graph("my_graph_flow", build_graph)

mcp_server(name: str, tools: List[str] = [])

Register MCP server metadata.

mcp_server("my_server", tools=["tool1", "tool2"])

Facade Imports

Pydantic

from aiel_core.pydantic import BaseModel, Field

class MyModel(BaseModel):
    field: str = Field(..., description="Description")

LangGraph

from aiel_core.langgraph.graph import StateGraph, START, END, MessagesState

graph = StateGraph(MessagesState)

LangChain Core

from aiel_core.langchain.core import ChatPromptTemplate, PromptTemplate, Runnable, RunnableConfig

template = ChatPromptTemplate.from_template("Hello {name}")

LangSmith

from aiel_core.langsmith.client import Client

client = Client()

---

Runtime Execution

Execution Plane Architecture

In production, the Execution Plane runs your code using aiel-runtime:

1. Download: Fetches project snapshot from Data Plane
2. Import: Loads entry_point.py to register exports
3. Validate: Ensures contracts are properly implemented
4. Execute: Invokes exports in a sandboxed runtime

The runtime is invoked via: python -m aiel_runtime.runner

Runtime Bundles

The Execution Plane provides curated runtime images with approved dependencies:

Minimal Runtime

aiel-runtime==X.Y.Z
aiel-sdk==X.Y.Z

Core Python dependencies only.

AI Runtime

aiel-runtime[ai]==X.Y.Z

Includes curated orchestration dependencies:

• langgraph
• langchain-core
• langsmith (optional)

Data Plane Manifest

Projects include a manifest specifying:

• runtime: Which runtime bundle/image to use
• sdk_version: Target SDK contract level

The Execution Plane allowlists runtime identifiers and pins package versions accordingly.

---

Best Practices

Project Structure

my-aiel-project/
├── entry_point.py          # Main export definitions
├── tools/
│   ├── __init__.py
│   └── validation.py       # Tool implementations
├── agents/
│   ├── __init__.py
│   └── orchestration.py    # Agent implementations
├── flows/
│   ├── __init__.py
│   └── main.py            # Flow definitions
├── models/
│   ├── __init__.py
│   └── schemas.py         # Pydantic models
├── requirements.txt
└── README.md

Type Safety

Always use Pydantic models for input validation:

from aiel_core.pydantic import BaseModel, Field

class Input(BaseModel):
    field: str = Field(..., min_length=1)

@tool("my_tool")
def my_tool(ctx, payload: dict) -> dict:
    data = Input.model_validate(payload)
    # Type-safe access
    return {"result": data.field}

Error Handling

Implement proper error handling and logging:

@tool("safe_tool")
def safe_tool(ctx, payload: dict) -> dict:
    try:
        # Process data
        ctx.log("processing", status="started")
        result = process(payload)
        ctx.log("processing", status="completed")
        return result
    except Exception as e:
        ctx.log("error", message=str(e))
        return {"error": str(e), "status": "failed"}

Naming Conventions

• Use snake_case for function and variable names
• Use descriptive names that reflect functionality
• Prefix internal helpers with underscore: _internal_helper()

---

Troubleshooting

Common Issues

Missing Integration Error

Error: ImportError: aiel.langgraph is not available

Solution: Install the required extra:

pip install "aiel-sdk[langgraph]"

Contract Validation Failure

Error: Contract validation failed: invalid signature

Solution: Ensure your functions match the expected signature:

• Tools: (ctx, payload: dict) -> dict
• Agents: (ctx, state: dict) -> dict
• Flows: (ctx, input: dict) -> dict
• HTTP handlers: (ctx, body/query: dict) -> dict

Import Path Issues

Error: ModuleNotFoundError: No module named 'aiel'

Solution: Use correct import paths:

• SDK exports: from aiel_sdk import tool, agent
• Facade imports: from aiel_core.langgraph.graph import StateGraph

Getting Help

• Documentation: [Internal docs portal]
• Support: [Internal support channel]
• Issues: [Internal issue tracker]

---

Migration Guide

From Legacy SDK

If migrating from a previous SDK version:

1. Update imports to use aiel.* facade
2. Update decorator signatures to match new contracts
3. Replace direct library imports with facade imports
4. Update manifest sdk_version field

---

Contributing

[Internal contribution guidelines]

---

License

[Internal license information]

---

Changelog

See CHANGELOG.md for version history.

---

Maintained by: Platform Engineering Team
Last Updated: January 2026

Local dev from aiel_sdk.memory import InMemorySaver, ThreadMessage

saver = InMemorySaver() saver.thread_write(thread_id="t1", messages=[ThreadMessage(role="user", content="hi")]) print(saver.thread_read(thread_id="t1").messages[-1].content) Production (remote) from aiel_sdk import AielClient from aiel_sdk.memory import RemoteSaver, ThreadMessage

client = AielClient(base_url="http://...", api_key="...")

workspace/project come from your Context; you bind once

saver = RemoteSaver(client, workspace_id=ctx.workspace_id, project_id=ctx.project_id)

saver.thread_write(thread_id="t1", messages=[ThreadMessage(role="user", content="hi")]) state = saver.state_patch(thread_id="t1", patch={"step": "triage"}) cp = saver.checkpoint_write(thread_id="t1", state={"step": "triage"}, metadata={"node": "triage_agent"})