aiel-sdk

AIEL SDK — Contract Layer for Governed AI Execution

aiel-sdk is the local development contract for the AI Execution Layer.

It provides the stable Python interface used to define tools, agents, flows, HTTP handlers, MCP servers, integration calls, and memory operations that can later execute inside the AIEL Execution Plane.

The SDK does not replace the runtime.

The SDK defines what can be executed.

The Execution Plane decides how, where, and under which policies it is executed.

---

Why the SDK Exists

AI applications are easy to prototype locally.

They become difficult to operate when they need to:

• call external systems safely
• expose tools to agents
• run under policy control
• preserve execution traceability
• execute consistently across environments
• support runtime-managed integrations
• share memory across workflows
• move from local development to production without rewriting code

The AIEL SDK solves this by giving developers a contract-first programming model.

Developers define executable capabilities locally.

The platform packages, validates, registers, and executes those capabilities remotely through aiel-runtime.

---

Position in the Platform

AIEL is composed of multiple layers.

The SDK is the developer-facing contract layer.

It is used before runtime execution, but its definitions become the runtime execution surface.

---

Execution Model

The SDK is designed around explicit exports.

An export is a capability that can be discovered, validated, registered, and executed by the platform.

Supported export types:

Exports are registered through decorators.

When entry_point.py is imported, the SDK registers those exports into an in-memory registry.

The runtime later uses the same registry contract to discover executable capabilities.

---

Architecture Flow

┌──────────────────────┐
│ Developer Code       │
│ entry_point.py       │
└──────────┬───────────┘
           │
           │ decorators
           ▼
┌──────────────────────┐
│ AIEL SDK Registry    │
│ tools / agents /     │
│ routers / flows /    │
│ http / mcp           │
└──────────┬───────────┘
           │
           │ validate_contracts()
           ▼
┌──────────────────────┐
│ Contract Validation  │
└──────────┬───────────┘
           │
           │ aiel push
           ▼
┌──────────────────────┐
│ Snapshot Packaging   │
└──────────┬───────────┘
           │
           ▼
┌─────────────────────────────┐
│ Execution Plane             │
├─────────────────────────────┤
│ Runtime Hydration           │
│ Registry Loading            │
│ Policy Enforcement          │
│ Integration Resolution      │
│ Memory Access               │
│ Execution Routing           │
│ Trace Collection            │
└──────────┬──────────────────┘
           │
           ▼
┌─────────────────────────────┐
│ Governed Runtime Execution  │
└─────────────────────────────┘

---

Local vs Runtime Responsibilities

The SDK keeps local development aligned with production execution.

---

Installation

Minimal Installation

Use this for core decorators, registry, contracts, client primitives, and local development.

pip install aiel-sdk

Full Development Installation

Use this when developing with curated optional integrations.

pip install "aiel-sdk[all]"

The all extra includes curated support for:

• LangGraph
• LangChain Core
• LangSmith
• AIEL CLI helper facades

Local Repository Installation

From a repository checkout:

pip install -e .

With all optional dependencies:

pip install -e ".[all]"

For zsh, always quote extras:

pip install -e ".[all]"

---

Requirements

• Python 3.10+
• pip 21+
• AIEL API token for remote API operations
• Workspace and project scope for governed integrations and memory

---

Quick Start

Create an entry_point.py.

The runtime imports this file to discover executable exports.

from aiel_sdk import tool, agent, router, flow, http, mcp_server
from aiel_sdk.context import Context

@tool("normalize_email")
def normalize_email(ctx: Context, payload: dict) -> dict:
    email = str(payload.get("email") or "").strip().lower()
    ctx.log("normalize_email", email=email)
    return {"email": email}

@agent("collect_personal_data")
def collect_personal_data(ctx: Context, state: dict) -> dict:
    out = normalize_email(ctx, {"email": state.get("email")})
    return {**dict(state), **out}

@router("next_step")
def next_step(ctx: Context, state: dict) -> str:
    return "done" if state.get("email") else "collect_personal_data"

@flow("driver_onboarding")
def driver_onboarding(ctx: Context, input: dict) -> dict:
    state = collect_personal_data(ctx, input)
    return {"status": "ok", "state": state}

@http.post("/driver/onboard")
def http_driver_onboard(ctx: Context, body: dict) -> dict:
    return driver_onboarding(ctx, body)

mcp_server("driver_support", tools=["normalize_email"])

---

Validate Contracts Locally

Importing entry_point.py registers all exports.

from aiel_sdk import get_registry, reset_registry, validate_contracts

import entry_point

validate_contracts()

reg = get_registry()

print("tools:", sorted(reg.tools.keys()))
print("agents:", sorted(reg.agents.keys()))
print("routers:", sorted(reg.routers.keys()))
print("flows:", sorted(reg.flows.keys()))
print("http:", [(h.method, h.path) for h in reg.http_handlers])
print("mcp servers:", sorted(reg.mcp_servers.keys()))

reset_registry()

Use reset_registry() between test runs when multiple import cycles are expected.

---

Registry Model

The registry is the SDK’s local source of truth for exported capabilities.

When decorators are evaluated, the SDK records metadata about each export.

The registry supports:

• export discovery
• local inspection
• contract validation
• runtime loading
• deployment packaging
• execution surface generation

Supported registry groups:

---

Decorator Contracts

Tool

A tool is a discrete action.

@tool("normalize_email")
def normalize_email(ctx: Context, payload: dict) -> dict:
    return {"email": str(payload["email"]).strip().lower()}

Recommended signature:

(ctx: Context, payload: dict) -> dict

Use tools for:

• validation
• transformation
• integration calls
• deterministic actions
• reusable execution units

---

Agent

An agent is a stateful reasoning or orchestration step.

@agent("collect_personal_data")
def collect_personal_data(ctx: Context, state: dict) -> dict:
    out = normalize_email(ctx, {"email": state.get("email")})
    return {**state, **out}

Recommended signature:

(ctx: Context, state: dict) -> dict

Use agents for:

• state transitions
• multi-step orchestration
• decision support
• interaction with tools

---

Router

A router returns the next execution path.

@router("next_step")
def next_step(ctx: Context, state: dict) -> str:
    return "done" if state.get("email") else "collect_personal_data"

Recommended signature:

(ctx: Context, state: dict) -> str

Use routers for:

• branching logic
• workflow transitions
• conditional execution
• agent selection

---

Flow

A flow is a top-level workflow entry point.

@flow("driver_onboarding")
def driver_onboarding(ctx: Context, input: dict) -> dict:
    state = collect_personal_data(ctx, input)
    return {"status": "ok", "state": state}

Recommended signature:

(ctx: Context, input: dict) -> dict

Use flows for:

• public workflow entry points
• orchestration units
• runtime-invokable business workflows

---

HTTP Handler

HTTP handlers expose flows or tools through runtime-managed HTTP routes.

@http.post("/driver/onboard")
def http_driver_onboard(ctx: Context, body: dict) -> dict:
    return driver_onboarding(ctx, body)

Supported pattern:

@http.get("/path")
@http.post("/path")

Use HTTP handlers when:

• external callers need an HTTP interface
• workflows need to be exposed as managed endpoints
• the Execution Plane should own endpoint routing

---

MCP Server

MCP server metadata declares which tools should be exposed through an MCP-compatible surface.

mcp_server("driver_support", tools=["normalize_email"])

Use MCP metadata when:

• tools should be discoverable by MCP-compatible clients
• tool access must remain governed by platform policies
• tool schemas should be exposed without custom server plumbing

---

Context Object

Every executable export receives a Context.

The context provides the runtime boundary between user code and platform services.

Typical context responsibilities include:

• structured logging
• execution metadata access
• request correlation
• workspace/project awareness
• runtime utilities
• policy-aware service access

Example:

@tool("normalize_email")
def normalize_email(ctx: Context, payload: dict) -> dict:
    email = str(payload.get("email") or "").strip().lower()
    ctx.log("normalize_email", email=email)
    return {"email": email}

---

AielClient

AielClient is the SDK client for calling AIEL APIs.

It is used for:

• resolving workspace/project scope
• calling governed integrations
• accessing remote memory
• invoking platform APIs
• using typed wrappers around remote capabilities

Authentication Resolution

AielClient resolves credentials in this order:

1. api_key= passed directly
2. AIEL_TOKEN environment variable
3. OS keyring entry ("aiel", "default"), when available

Example:

from aiel_sdk import AielClient

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

workspace_id, project_id = client.resolve_scope()

Workspace and project can also be resolved from:

AIEL_WORKSPACE_ID
AIEL_PROJECT_ID

---

Integration Access

Integrations are accessed through governed platform connections.

Application code should not embed raw third-party credentials.

Instead, it references platform-managed connection IDs.

This allows the platform to enforce:

• workspace scope
• project scope
• provider policy
• allowed actions
• audit logging
• credential isolation

---

Generic Integration API

from aiel_sdk import AielClient

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

connections = client.integrations.list("ws_123", "proj_456")

connection = connections[0]

detail = client.integrations.get(
    connection.workspace_id,
    connection.project_id,
    connection.connection_id,
)

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

---

Typed Integration Wrappers

Typed wrappers provide provider-specific request and response models on top of governed platform connections.

Slack

from aiel_sdk import AielClient
from aiel_sdk.integrations.slack import RemoteSlack, SlackPostMessageRequest

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

slack = RemoteSlack(client, connection_id="conn_slack_123")

slack.post_message(
    SlackPostMessageRequest(
        channel="#general",
        text="Hello from AIEL SDK",
    )
)

OpenAI

from aiel_sdk import AielClient
from aiel_sdk.integrations.openai import RemoteOpenAI, OpenAIResponsesCreateRequest

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

oai = RemoteOpenAI(client, connection_id="conn_openai_123")

result = oai.responses_create(
    OpenAIResponsesCreateRequest(
        input="Say hello",
        model="gpt-4o-mini",
    )
)

print(result.raw)

---

Memory

The SDK supports both local and remote memory implementations.

Use local memory for development and tests.

Use remote memory for production workflows running through AIEL-managed workspace/project scope.

---

Local Memory

from aiel_sdk.memory import InMemorySaver, ThreadMessage

saver = InMemorySaver()

saver.thread_write(
    thread_key="t1",
    messages=[
        ThreadMessage(role="human", content="hi"),
    ],
)

thread = saver.thread_read(thread_key="t1")

print(thread.messages[-1].content)

Local memory:

• does not require network access
• is useful for tests
• is not durable
• should not be used as production state

---

Remote Memory

from aiel_sdk import AielClient
from aiel_sdk.memory import RemoteSaver, ThreadMessage, RecallQuery

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

remote = RemoteSaver(
    client,
    workspace_id="ws_123",
    project_id="proj_456",
)

remote.thread_write(
    thread_key="t1",
    messages=[
        ThreadMessage(role="human", content="hi"),
    ],
    ttl_seconds=3600,
)

state = remote.state_patch(
    thread_key="t1",
    patch={"step": "triage"},
    ttl_seconds=3600,
)

checkpoint = remote.checkpoint_write(
    thread_key="t1",
    state=state.state,
    metadata={"node": "triage_agent"},
    ttl_seconds=86400,
)

items = remote.recall_search(
    query=RecallQuery(
        namespace=["users", "u1"],
        limit=10,
    )
)

Remote memory supports:

• thread history
• state patches
• checkpoints
• recall search
• TTL-based lifecycle control
• workspace/project isolation

---

Facade Modules

The SDK provides curated facade modules for approved runtime dependencies.

Facade imports are used to keep local development aligned with managed runtime environments.

Example:

from aiel_core.pydantic import BaseModel, Field

class NormalizeEmailPayload(BaseModel):
    email: str = Field(..., description="Raw email address")

LangGraph example:

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

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": []})

Missing optional dependencies fail with actionable errors that identify the required installation extra.

---

Runtime Execution

In production, application code is executed by aiel-runtime inside the Execution Plane.

The runtime performs the following sequence:

1. Fetches the project snapshot from the Data Plane
2. Hydrates the approved runtime environment
3. Imports entry_point.py
4. Loads the SDK registry
5. Validates registered contracts
6. Resolves workspace/project scope
7. Applies integration and policy controls
8. Invokes the requested export
9. Captures logs, traces, and execution metadata
10. Returns a normalized runtime response

Runtime invocation is handled by the platform.

Local code should be written against the SDK contract, not against runtime internals.

---

SDK to Execution Plane Flow

Developer
   │
   │ writes entry_point.py
   ▼
AIEL SDK
   │
   │ decorators register exports
   ▼
Local Registry
   │
   │ validate_contracts()
   ▼
AIEL CLI
   │
   │ aiel push
   ▼
Snapshot
   │
   │ versioned + immutable
   ▼
Execution Plane
   │
   │ aiel-runtime loads registry
   ▼
Governed Execution
   │
   ├── policies
   ├── integrations
   ├── memory
   ├── traces
   └── audit logs

---

Runtime Bundles

The Execution Plane may provide curated runtime bundles with approved dependency sets.

Minimal Runtime

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

Used for:

• core tools
• flows
• routers
• HTTP handlers
• basic integrations

AI Runtime

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

May include curated AI orchestration dependencies such as:

• LangGraph
• LangChain Core
• LangSmith

Runtime identifiers should be allowlisted by the platform.

Package versions should be pinned by runtime bundle.

---

Data Plane Manifest

Project snapshots should include manifest metadata such as:

The manifest is used by the Execution Plane to determine how the snapshot should be loaded and executed.

---

Recommended Project Structure

my-aiel-project/
├── entry_point.py
├── tools/
│   ├── __init__.py
│   └── validation.py
├── agents/
│   ├── __init__.py
│   └── onboarding.py
├── flows/
│   ├── __init__.py
│   └── driver_onboarding.py
├── models/
│   ├── __init__.py
│   └── schemas.py
├── tests/
│   └── test_contracts.py
├── requirements.txt
└── README.md

---

Type Safety

Use Pydantic models for validation at the boundary of tools, agents, and flows.

from aiel_core.pydantic import BaseModel, Field
from aiel_sdk import tool
from aiel_sdk.context import Context

class NormalizeEmailPayload(BaseModel):
    email: str = Field(..., description="Raw email address")

class NormalizeEmailResult(BaseModel):
    email: str

@tool("normalize_email")
def normalize_email(ctx: Context, payload: dict) -> dict:
    data = NormalizeEmailPayload.model_validate(payload)
    email = data.email.strip().lower()
    return NormalizeEmailResult(email=email).model_dump()

Recommended practice:

• validate external input
• return JSON-serializable dictionaries
• keep tool outputs explicit
• use typed request models for integrations
• avoid hidden global state

---

Logging

Use ctx.log(...) for structured runtime logs.

@tool("normalize_email")
def normalize_email(ctx: Context, payload: dict) -> dict:
    email = str(payload.get("email") or "").strip().lower()
    ctx.log("normalize_email.completed", email=email)
    return {"email": email}

Logs should be:

• structured
• safe to persist
• free of raw secrets
• useful for trace correlation

---

Error Handling

Prefer explicit errors and structured return values.

@tool("safe_normalize_email")
def safe_normalize_email(ctx: Context, payload: dict) -> dict:
    try:
        email = str(payload.get("email") or "").strip().lower()

        if not email:
            return {
                "status": "invalid",
                "reason": "email_required",
            }

        return {
            "status": "ok",
            "email": email,
        }

    except Exception as exc:
        ctx.log("safe_normalize_email.failed", error=str(exc))
        return {
            "status": "failed",
            "reason": "unexpected_error",
        }

Avoid leaking:

• access tokens
• provider credentials
• customer secrets
• raw personal data
• internal stack traces in user-facing outputs

---

Testing

Run Unit Tests

python -m unittest discover -s tests -v

Run SDK Smoke Example

From a repository checkout:

PYTHONPATH=src python -m aiel_sdk.examples.smoke

Example Contract Test

def test_contracts_are_valid():
    import entry_point

    from aiel_sdk import get_registry, validate_contracts

    validate_contracts()

    reg = get_registry()

    assert "normalize_email" in reg.tools
    assert "driver_onboarding" in reg.flows

---

Best Practices

Keep entry_point.py Small

Use entry_point.py as the registration surface.

Move business logic into modules.

Recommended:

# entry_point.py
from aiel_sdk import tool
from tools.email import normalize_email_impl

@tool("normalize_email")
def normalize_email(ctx, payload):
    return normalize_email_impl(ctx, payload)

---

Use Platform-Managed Integrations

Do not store provider credentials in application code.

Use AIEL connection IDs and governed integration wrappers.

Recommended:

slack = RemoteSlack(client, connection_id="conn_slack_123")

Avoid:

# Do not do this
slack_token = "xoxb-..."

---

Separate Local and Remote State

Use InMemorySaver for tests.

Use RemoteSaver for production.

Do not assume local memory behavior matches durable remote state semantics.

---

Keep Export Names Stable

Export names become part of the execution contract.

Changing names can affect:

• invocation references
• runtime routing
• MCP exposure
• deployment compatibility
• external callers

Recommended:

@flow("driver_onboarding")

Avoid frequently renaming exports after they are used by production callers.

---

Make Outputs Explicit

Return structured dictionaries with clear status fields.

Recommended:

return {
    "status": "ok",
    "state": state,
}

Avoid ambiguous responses:

return state

---

Security Model

The SDK supports the platform security model by avoiding direct credential ownership in application code.

Security boundaries are enforced by the platform, including:

• workspace scope
• project scope
• connection scope
• action-level permissions
• policy evaluation
• audit logging
• runtime isolation

The SDK should be treated as the local contract layer, not the policy enforcement engine.

Policy enforcement occurs in the managed platform and runtime environment.

---

Operational Guarantees

When used with AIEL CLI and AIEL Runtime, the SDK enables:

• explicit execution contracts
• discoverable capabilities
• predictable runtime loading
• typed local development
• governed integration access
• workspace/project-scoped memory
• runtime traceability
• version-aware deployment
• consistent local-to-production imports

---

Troubleshooting

Missing Optional Dependency

Example error:

ImportError: aiel_core.langgraph is not available

Install the required extra:

pip install "aiel-sdk[langgraph]"

Or install all curated integrations:

pip install "aiel-sdk[all]"

---

Contract Validation Failure

Typical cause:

• invalid function signature
• missing context argument
• unsupported return shape
• decorator registered against the wrong callable

Check signatures:

---

Registry Is Empty

Possible causes:

• entry_point.py was not imported
• decorators are inside a function that was not called
• exports are defined behind conditional logic
• registry was reset before inspection

Fix:

import entry_point

from aiel_sdk import get_registry

reg = get_registry()
print(reg.tools.keys())

---

Authentication Failure

AielClient could not resolve credentials.

Check one of the following:

• pass api_key=...
• set AIEL_TOKEN
• authenticate through AIEL CLI if keyring integration is available

---

Workspace or Project Not Resolved

Set scope explicitly:

client = AielClient(
    base_url="https://api.example.com",
    api_key="REPLACE_ME",
    workspace_id="ws_123",
    project_id="proj_456",
)

Or use environment variables:

export AIEL_WORKSPACE_ID=ws_123
export AIEL_PROJECT_ID=proj_456

---

Migration Guidance

When moving existing AI application code to AIEL:

1. Identify reusable actions and define them as tools.
2. Identify stateful reasoning steps and define them as agents.
3. Identify top-level workflows and define them as flows.
4. Expose external endpoints through @http.
5. Move provider credentials into platform-managed integrations.
6. Replace direct third-party calls with governed integration wrappers.
7. Add local contract validation tests.
8. Package and deploy through aiel-cli.
9. Execute through the AIEL Execution Plane.

---

Relationship to the CLI

The SDK defines the contract.

The CLI deploys the contract.

Typical workflow:

pip install aiel-sdk
pip install aiel-cli

aiel auth login
aiel config set workspace <workspace-slug>
aiel config set project <project-slug>
aiel init
aiel add .
aiel commit -m "Add driver onboarding flow"
aiel push

---

Relationship to the Runtime

The runtime executes the contract.

At runtime:

• entry_point.py is imported
• decorators register exports
• contracts are validated
• requested exports are invoked
• integrations and memory are resolved through platform services
• logs and traces are captured

Developers should not depend on runtime internals directly.

Use SDK interfaces for local development and runtime compatibility.

---

Changelog

Version history is managed through Git tags and PyPI releases.

---

License

See LICENSE.

---

Summary

aiel-sdk is the contract layer for governed AI execution.

It lets teams define tools, agents, flows, HTTP handlers, MCP surfaces, integration calls, and memory operations locally while keeping production execution under the control of the AIEL Execution Plane.

The SDK gives developers a stable programming model.

The platform provides the runtime, governance, isolation, and operational controls required for production AI systems.