enki-py 0.5.75

Python bindings for Enki's Rust agent runtime.

enki-py

Python bindings for Enki's Rust agent runtime.

enki-py exposes two layers:

• A high-level Python API built around Agent and MultiAgentRuntime
• The lower-level native bindings generated from the Rust runtime, including workflow APIs

The package is designed for Python-first usage without giving up the native runtime features that already exist in the Rust core.

Requirements

• Python >=3.8
• A supported model string such as openai::gpt-4o, anthropic::claude-sonnet-4-6, or ollama::qwen3.5:latest

Install

Published package:

pip install enki-py

With the built-in LiteLLM adapter:

pip install "enki-py[litellm]"

Using uv:

uv add enki-py
uv add "enki-py[litellm]"

What It Exports

High-level Python API

The main Python-facing exports are:

• Agent
• AgentLoopRequest
• AgentLoopResult
• AgentRunResult
• ExecutionStep
• Tool
• ToolRegistry
• RunContext
• MemoryBackend
• MemoryModule
• MemoryEntry
• MemoryKind
• MultiAgentRuntime
• MultiAgentMember
• AgentCard
• LiteLlmProvider
• LlmProviderBackend

Low-level native bindings

The generated module is also re-exported, so lower-level runtime types remain available when you want to work closer to the Rust layer. Commonly used native types include:

• EnkiAgent
• EnkiWorkflowRuntime
• EnkiTool and EnkiToolSpec
• EnkiMemoryEntry
• EnkiMemoryModule

Agent API

Agent is the main Python entrypoint. Its constructor supports:

• model: required provider and model string
• deps_type: optional dependency type used with RunContext
• instructions: system guidance / prompt preamble
• agentic_loop: optional prompt-level loop instructions
• agent_loop_handler: optional Python callback that overrides the runtime loop
• name: display name for the agent
• max_iterations: iteration cap for the runtime loop
• workspace_home: optional workspace root for persisted runtime state
• tools: optional list of pre-registered Tool objects
• tool_registry: optional reusable ToolRegistry
• memories: optional list of MemoryModule objects
• llm: optional Python-side LLM backend or callback

Main methods:

• run(...): async single-agent execution
• run_sync(...): sync wrapper around run(...)
• set_agent_loop_handler(...)
• clear_agent_loop_handler(...)
• tool_plain(...): decorator for plain Python tools
• tool(...): decorator for tools that receive RunContext
• register_tool(...)
• connect_tool_registry(...)
• register_memory(...)
• as_workflow_agent(...): converts the wrapper into a workflow-configured low-level agent

The result of run(...) or run_sync(...) is AgentRunResult, which contains:

• output: final model response text
• steps: ExecutionStep entries describing runtime progress

Both run(...) and run_sync(...) accept:

• session_id: optional explicit session id
• deps: optional dependency object passed into RunContext
• on_step: optional callback for streaming step visibility

Basic Agent

Synchronous example:

import uuid

from enki_py import Agent


agent = Agent(
    "anthropic::claude-sonnet-4-6",
    name="Simple Agent",
    instructions="Answer clearly and keep responses short.",
)

result = agent.run_sync(
    "Explain what this Enki Python example demonstrates.",
    session_id=f"simple-agent-{uuid.uuid4()}",
)

print(result.output)
for step in result.steps:
    print(f"{step.index}. [{step.phase}] {step.kind}: {step.detail}")

Async version:

import asyncio

from enki_py import Agent


async def main() -> None:
    agent = Agent(
        "ollama::qwen3.5:latest",
        name="Async Agent",
        instructions="Answer clearly and keep responses short.",
    )

    result = await agent.run("What does enki-py provide?")
    print(result.output)


asyncio.run(main())

Tools

The high-level wrapper lets you register Python functions directly as tools.

@agent.tool_plain

Use this when the tool only needs its declared arguments:

from enki_py import Agent


agent = Agent(
    "ollama::qwen3.5:latest",
    name="Researcher",
    instructions="Use tools when they help answer factual questions.",
)


@agent.tool_plain
def lookup_example_topics(topic: str) -> str:
    """Return a canned fact for an example topic."""
    facts = {
        "memory": "Memory lets the agent persist and recall useful session context.",
        "tools": "Tools let the agent call Python functions to fetch or compute structured results.",
        "multi-agent": "Multi-agent runtimes let a coordinator route work to specialized agents.",
    }
    return facts.get(topic.lower(), f"No prepared fact exists for '{topic}'.")

@agent.tool

Use this when the tool should receive a RunContext with typed dependencies:

from dataclasses import dataclass

from enki_py import Agent, RunContext


@dataclass
class AppDeps:
    project_name: str


agent = Agent(
    "ollama::qwen3.5:latest",
    deps_type=AppDeps,
    name="Context Agent",
    instructions="Use the provided context when it helps answer.",
)


@agent.tool
def project_name(ctx: RunContext[AppDeps]) -> str:
    """Return the active project name from runtime dependencies."""
    return ctx.deps.project_name

Tool schemas are inferred from Python signatures and type annotations. Required parameters come from arguments without defaults, and optional parameters come from arguments with defaults.

If you need lower-level control, you can also build Tool(...) values directly and register them with register_tool(...).

Reusable ToolRegistry

If you want to manage a shared set of tools separately from a single agent, build a ToolRegistry and connect it when needed:

from enki_py import Agent, ToolRegistry


registry = ToolRegistry()


@registry.tool_plain
def lookup_release(version: str) -> str:
    """Return a canned release note."""
    return f"Release {version} is ready."


agent = Agent(
    "ollama::qwen3.5:latest",
    name="Registry Agent",
    instructions="Use connected tools when they help.",
)

agent.connect_tool_registry(registry)

You can also pass tool_registry=registry to Agent(...) during construction.

This is the cleanest path when multiple agents should share the same tool catalog or when you want to prepare tools separately from agent creation.

Memory

enki-py supports Python-defined memory modules through MemoryBackend and MemoryModule.

Implement MemoryBackend when you want custom record and recall behavior:

from dataclasses import dataclass, field
from time import time_ns

from enki_py import MemoryBackend, MemoryEntry, MemoryKind


@dataclass
class SessionMemory(MemoryBackend):
    name: str = "session_memory"
    _entries: dict[str, list[MemoryEntry]] = field(default_factory=dict)

    def record(self, session_id: str, user_msg: str, assistant_msg: str) -> None:
        entries = self._entries.setdefault(session_id, [])
        entries.append(
            MemoryEntry(
                key=f"{session_id}-{len(entries)}",
                content=f"User: {user_msg}\nAssistant: {assistant_msg}",
                kind=MemoryKind.RECENT_MESSAGE,
                relevance=1.0,
                timestamp_ns=time_ns(),
            )
        )

    def recall(self, session_id: str, query: str, max_entries: int) -> list[MemoryEntry]:
        entries = self._entries.get(session_id, [])
        return entries[-max_entries:]

    def flush(self, session_id: str) -> None:
        return None

Register it like this:

memory = SessionMemory()
agent = Agent(
    "ollama::qwen3.5:latest",
    memories=[memory.as_memory_module()],
)

Supported memory kinds:

• MemoryKind.RECENT_MESSAGE
• MemoryKind.SUMMARY
• MemoryKind.ENTITY
• MemoryKind.PREFERENCE

Memory callbacks may be synchronous or async def coroutines.

Multi-Agent Runtime

MultiAgentRuntime wires multiple Agent instances together and installs the coordination tools used for agent discovery and delegation.

Available methods:

• registry()
• discover(...)
• process(...)
• process_sync(...)

The runtime automatically installs two reserved tools on member agents:

• discover_agents
• delegate_task

Minimal example:

from enki_py import Agent, MultiAgentMember, MultiAgentRuntime


coordinator = Agent(
    "ollama::qwen3.5:latest",
    name="Coordinator",
    instructions=(
        "Use discover_agents first. "
        "Delegate research work to the researcher with delegate_task."
    ),
)

researcher = Agent(
    "ollama::qwen3.5:latest",
    name="Researcher",
    instructions="Answer delegated questions clearly and briefly.",
)

runtime = MultiAgentRuntime(
    [
        MultiAgentMember(
            agent_id="coordinator",
            agent=coordinator,
            capabilities=["planning", "orchestration"],
        ),
        MultiAgentMember(
            agent_id="researcher",
            agent=researcher,
            capabilities=["research"],
        ),
    ]
)

result = runtime.process_sync(
    "coordinator",
    "Use discover_agents first, then delegate to the researcher.",
    session_id="simple-multi-agent-example",
)

print(result.output)

Repository examples:

• example/enki-py/simple_multi_agent.py
• example/enki-py/multi_agent_with_memory_and_tools.py

Workflow Runtime

For workflow execution, the lower-level EnkiWorkflowRuntime is the main entrypoint. A common pattern is:

1. Create Python Agent wrappers
2. Convert them with as_workflow_agent(agent_id=..., capabilities=[...])
3. Provide JSON task definitions and JSON workflow definitions
4. Start and inspect runs through the workflow runtime

Methods used in the checked-in Python examples include:

• list_workflows_json()
• list_runs_json()
• inspect_json(run_id)
• start_json(payload_json)
• resume_json(run_id)
• submit_intervention_json(run_id, intervention_id, response)

Workflow example:

import asyncio
import json
from pathlib import Path

import enki_py


WORKSPACE_HOME = Path("./example/enki-py/.enki-workflow")
MODEL = "ollama::qwen3.5:latest"


async def main() -> None:
    researcher = enki_py.Agent(
        MODEL,
        name="Researcher",
        instructions="Return short factual notes.",
        workspace_home=str(WORKSPACE_HOME),
    )
    writer = enki_py.Agent(
        MODEL,
        name="Writer",
        instructions="Turn notes into a concise summary.",
        workspace_home=str(WORKSPACE_HOME),
    )

    runtime = enki_py.EnkiWorkflowRuntime(
        agents=[
            researcher.as_workflow_agent(agent_id="researcher", capabilities=["research"]),
            writer.as_workflow_agent(agent_id="writer", capabilities=["writing"]),
        ],
        tasks_json=[],
        workflows_json=[],
        workspace_home=str(WORKSPACE_HOME),
    )

    print(json.loads(await runtime.list_workflows_json()))


asyncio.run(main())

For a complete runnable version with task and edge definitions, see example/enki-py/agent_workflow.py.

Human Intervention

Workflow runs can pause for human input and then resume from persisted state.

Two built-in patterns are demonstrated in the repository examples:

• human_gate workflow nodes that pause and wait for a human response
• task nodes with failure_policy: "pause_for_intervention" that convert a failure into a human resolution step

The runtime interaction loop is:

1. start_json(...) starts the workflow and may return a paused response
2. inspect_json(run_id) exposes pending_interventions
3. submit_intervention_json(run_id, intervention_id, response) resolves the intervention
4. resume_json(run_id) continues the persisted run

See example/enki-py/human_intervention_workflow.py.

Custom LLM Providers

You can keep model execution on the Python side by passing either:

• an instance of LlmProviderBackend
• a compatible Python callable matching the LLM completion shape

The built-in LiteLlmProvider is the default adapter for LiteLLM-backed usage. It is useful when you want:

• provider-specific configuration in Python
• LiteLLM-managed credentials or routing
• a Python extension point without changing Rust code

Notes from the current implementation:

• LiteLlmProvider raises a clear error if litellm is not installed
• for Ollama, the default base URL comes from OLLAMA_URL or falls back to http://127.0.0.1:11434
• ENKI_LITELLM_TIMEOUT controls the request timeout used by the adapter
• Ollama tool calling is only forwarded when ENKI_OLLAMA_TOOLS is truthy

The repository workflow example also includes a custom OllamaProvider implementation in example/enki-py/agent_workflow.py.

Custom Agentic Loop

There are two different customization levels.

Use agentic_loop= when you want to keep the normal Rust runtime loop but replace the default loop instructions that the model sees:

from enki_py import Agent


agent = Agent(
    "anthropic::claude-sonnet-4-6",
    name="Custom Agentic Loop",
    instructions="Answer clearly and keep responses short.",
    agentic_loop=(
        "1. Understand the user request.\n"
        "2. Decide whether a tool is necessary before acting.\n"
        "3. If you use a tool, summarize what you learned.\n"
        "4. Verify that the answer is complete.\n"
        "5. Return the final response."
    ),
)

Use agent_loop_handler= when you want Python to own the turn-by-turn control flow:

from enki_py import Agent, AgentLoopRequest, AgentLoopResult, ExecutionStep


def custom_loop(request: AgentLoopRequest[None]) -> AgentLoopResult:
    return AgentLoopResult(
        output=f"Handled in Python for: {request.user_message}",
        steps=[
            ExecutionStep(
                index=1,
                phase="Custom",
                kind="final",
                detail="Returned a final answer from Python",
            )
        ],
    )


agent = Agent(
    "ollama::qwen3.5:latest",
    instructions="Answer clearly and keep responses short.",
    agent_loop_handler=custom_loop,
)

The loop request includes:

• session_id
• user_message
• system_prompt
• messages
• tools
• agent_dir
• workspace_dir
• sessions_dir
• model
• max_iterations
• deps

You can also update the loop handler after construction with:

• agent.set_agent_loop_handler(handler)
• agent.clear_agent_loop_handler()

Repository examples:

• example/enki-py/custom_agentic_loop.py
• example/enki-py/react_custom_agentic_loop.py
• example/enki-py/compare_agent_loops.py

Running The Examples

After building the local package with maturin develop, you can run the checked-in examples from the repository root:

python example\enki-py\simple_agent.py
python example\enki-py\simple_agent_ollama.py
python example\enki-py\tool_registry.py
python example\enki-py\simple_multi_agent.py
python example\enki-py\multi_agent_with_memory_and_tools.py
python example\enki-py\agent_workflow.py
python example\enki-py\human_intervention_workflow.py
python example\enki-py\custom_agentic_loop.py
python example\enki-py\react_custom_agentic_loop.py
python example\enki-py\compare_agent_loops.py

Model notes:

• simple_agent.py uses anthropic::claude-sonnet-4-6
• simple_agent_ollama.py uses ollama::qwen3.5
• several other checked-in examples use ollama::qwen3.5:latest
• agent_workflow.py reads ENKI_MODEL and falls back to ollama::qwen3.5:latest

Make sure the selected provider and model are available in your local environment before running the examples.

Development

From crates/bindings/enki-py:

pip install maturin
maturin develop
pytest python/tests

Useful commands:

• maturin develop: build the Rust extension and install the package in editable form
• pytest python/tests: run the Python-side tests and examples under test coverage

Project Layout

crates/bindings/enki-py/
|-- Cargo.toml
|-- pyproject.toml
|-- python/enki_py/
|-- python/tests/
`-- src/

• src/: Rust UniFFI binding implementation
• python/enki_py/: Python wrapper and generated native module
• python/tests/: Python-side tests and usage coverage

Related Docs

• Repository README
• Enki docs
• Getting started