aiel-sdk 1.3.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
• 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.

---

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"})