API Reference

Complete reference for the EVOID public API.

Core Imports

from evoid import (
    Intent, Level,
    register, resolve, all_intents, clear_registry,
    PipelineConfig, resolve_pipeline,
    Result, execute_pipeline,
    Context, fork,
    Processor, register_processor, get_processor, all_processors,
    Config, execute, execute_by_name,
    Message, subscribe, unsubscribe, publish, get_history,
    Service,
)

Intent

@dataclass(frozen=True)
class Intent:
    name: str
    level: Level = Level.STANDARD
    metadata: dict[str, Any] = field(default_factory=dict)
    timeout: float | None = None
    priority: int = 0

Frozen dataclass. Pure data. The runtime reads it, never mutates it.

Fields:

Level

class Level(str, Enum):
    EPHEMERAL = "ephemeral"
    STANDARD = "standard"
    CRITICAL = "critical"

Each level maps to a default pipeline and timeout:

Context

@dataclass
class Context:
    intent: Intent
    state: dict[str, Any] = field(default_factory=dict)
    deps: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)
    errors: list[Exception] = field(default_factory=list)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    created_at: datetime = field(default_factory=...)

Mutable execution context passed to every processor.

Functions:

fork(ctx: Context) -> Context

Create a child context with the same intent and deps, copied state, and parent_id in metadata.

Result

@dataclass(frozen=True)
class Result:
    success: bool
    value: Any = None
    error: Exception | None = None
    processors: tuple[str, ...] = ()
    duration: float = 0.0

Returned by every pipeline execution.

PipelineConfig

@dataclass(frozen=True)
class PipelineConfig:
    processors: tuple[str, ...] = ()
    priority: int = 0
    timeout: float | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

Describes which processors to run, in what order, with what constraints.

Registry Functions

register(intent: Intent) -> None

Register an intent definition in the global registry.

resolve(name: str) -> Intent | None

Look up an intent by name. Returns None if not found.

all_intents() -> dict[str, Intent]

Return a copy of all registered intents.

clear_registry() -> None

Remove all registered intents.

Processor Functions

Processor = Callable[[Context], Awaitable[Any]]

Type alias. A processor is any async function that takes a Context and returns a value.

register_processor(name: str, fn: Processor) -> None

Register a processor function by name.

get_processor(name: str) -> Processor | None

Look up a processor by name.

all_processors() -> dict[str, Processor]

Return a copy of all registered processors.

Runtime Functions

async execute(intent: Intent, config: Config | None = None) -> Result

Execute an intent through its resolved pipeline. This is the core function.

Resolution order:

1. Check for user overrides (extend.get_pipeline_config)
2. Fall back to default resolution (resolve_pipeline)

async execute_by_name(name: str, **kwargs) -> Result

Execute a registered intent by name. Extra kwargs are merged into the intent’s metadata.

@dataclass
class Config:
    name: str = "evoid-service"
    adapter: str = "asgi"
    engines: dict[str, str] = field(default_factory=dict)

Extend Functions

Add Intents

add_intent(intent: Intent, handler: Callable) -> None

Register an intent and its handler as a processor.

add_intent_with_pipeline(
    intent: Intent,
    processors: list[str],
    handler: Callable | None = None,
) -> None

Register an intent with a custom processor chain.

Modify Pipelines

before(intent_name: str, processor_name: str) -> None

Insert a processor at the start of the pipeline.

after(intent_name: str, processor_name: str) -> None

Append a processor at the end of the pipeline.

before_processor(intent_name: str, target: str, new: str) -> None

Insert before a specific processor.

after_processor(intent_name: str, target: str, new: str) -> None

Insert after a specific processor.

replace_pipeline(intent_name: str, processors: list[str]) -> None

Replace the entire pipeline.

remove_processor(intent_name: str, processor_name: str) -> None

Remove a processor from the pipeline.

Introspection

get_pipeline_config(intent: Intent) -> PipelineConfig

Get the effective pipeline config for an intent (overrides applied).

list_overrides() -> dict[str, list[str]]

Return all pipeline overrides as {intent_name: [processor_names]}.

clear_overrides() -> None

Remove all pipeline overrides.

Message Bus

@dataclass(frozen=True)
class Message:
    intent: Intent
    source: str = ""
    target: str = ""
    reply_to: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

Functions

subscribe(topic: str, handler: Callable[[Intent], Awaitable[Any]]) -> None

Subscribe to a topic. Topics match by intent name, level, or * (wildcard).

unsubscribe(topic: str, handler: Callable) -> bool

Remove a subscription. Returns True if found and removed.

async publish(intent: Intent, source: str = "", target: str = "") -> list[Any]

Publish an intent to all matching subscribers. Handlers run concurrently. Returns list of results.

get_history() -> list[Message]

Return a copy of all published messages (for debugging).

Parallel Execution

async gather(
    *intents: Intent,
    concurrency: int = 10,
    return_exceptions: bool = False,
) -> list[Result]

Execute multiple intents in parallel with a concurrency limit.

async gather_with_priority(
    *intents: Intent,
    concurrency: int = 10,
) -> list[Result]

Execute intents sorted by priority (higher first).

async parallel(
    *funcs: Callable[[], Awaitable[Any]],
    concurrency: int = 10,
) -> list[Any]

Execute arbitrary async functions in parallel.

run_in_thread(func, *args, **kwargs) -> Any

Run a synchronous function in a thread pool (blocking).

async run_in_thread_async(func, *args, **kwargs) -> Any

Run a synchronous function in a thread pool (async).

IntentQueue

class IntentQueue:
    def __init__(self, max_concurrent: int = 10) -> None: ...
    def enqueue(self, intent: Intent, priority: int = 0) -> None: ...
    def dequeue(self) -> Intent | None: ...
    async def process(self) -> list[Result]: ...
    @property
    def size(self) -> int: ...
    @property
    def is_empty(self) -> bool: ...

Priority queue for ordered intent execution.

Service (Core)

@dataclass
class Service:
    name: str
    handlers: dict[str, Handler] = field(default_factory=dict)
    running: bool = False

start(service: Service) -> None

Start a service. Registers all handlers with the message bus.

stop(service: Service) -> None

Stop a service.

on(service: Service, intent_name: str, handler: Handler) -> None

Register a handler for an intent. If the service is running, registers immediately with the message bus.

async call(service: Service, intent: Intent) -> Any

Call another service. Returns the first non-exception result.

async emit(service: Service, intent: Intent) -> list[Any]

Fire-and-forget. Publishes to all subscribers without waiting.

@route Style

from evoid.web.route import Service, get, post, put, delete

app = Service(name: str) -> App
get(path: str, level: str = "standard") -> Callable
post(path: str, level: str = "standard") -> Callable
put(path: str, level: str = "standard") -> Callable
delete(path: str, level: str = "standard") -> Callable

Extend functions:

before(route: str, processor: str) -> None
after(route: str, processor: str) -> None
before_handler(route: str, target: str, processor: str) -> None
after_handler(route: str, target: str, processor: str) -> None
replace_pipeline(route: str, processors: list[str]) -> None

Run:

async run(app: App, host: str = "0.0.0.0", port: int = 8000) -> None

@controller Style

from evoid.web.controller import Service, Controller, GET, POST, PUT, DELETE

app = Service(name: str) -> App
Controller(prefix: str = "", level: str = "standard") -> Callable
GET(path: str = "", level: str = "standard") -> Callable
POST(path: str = "", level: str = "standard") -> Callable
PUT(path: str = "", level: str = "standard") -> Callable
DELETE(path: str = "", level: str = "standard") -> Callable

Same extend functions and run() as @route.

Native Style

from evoid.native import create_service, on, run

create_service(name: str) -> Service
on(service: Service, intent: Intent, handler: Handler) -> None
async execute_service(service: Service, intent_name: str, **kwargs) -> Any
async run(service: Service, host: str = "0.0.0.0", port: int = 8000) -> None

CLI

evo init <name>              # Create new project
evo service new <name>       # Add service to project
evo service list             # List services
evo service run <name>       # Run a service
evo sync                     # Sync dependencies from evoid.toml
evo run                      # Run all services
evo serve [host] [port]      # Quick serve (single service)
evo list-intents             # List registered intents
evo list-processors          # List registered processors
evo exec <intent>            # Execute intent by name
evo version                  # Show version