autourgos-core

Stdlib-only shared utility library for the Autourgos framework — the concrete functions and classes that today are copy-pasted (identically or near-identically) across multiple unrelated autourgos-* package families, extracted once so a fix in one place fixes everywhere.

Status: all 6 planned modules implemented. See autourgos-core-architecture-plan.md for the full design and rollout plan (further extraction candidates in its Section 3.5 stay in their own base packages, not here).

Design principles

• Zero third-party dependencies. Standard library only — safe for even the most dependency-light packages in the framework to depend on.
• No domain knowledge. autourgos-core doesn't know what a memory message is, what an OpenAI response looks like, or what a Tkinter widget is. It provides primitives; domain shape stays in the package that owns that domain.
• Extraction only where 2+ unrelated package families already duplicate the same logic.

Modules

imports.py

from autourgos_core import try_import, require_available

available, modules, error = try_import("sounddevice")
require_available(available, f"'sounddevice' is required: {error}", ImportError)
sd = modules["sounddevice"]

text.py

from autourgos_core import extract_text, parse_param_descriptions

text = extract_text(llm.invoke("..."))  # handles both a plain string and a structured_output=True dict

descs = parse_param_descriptions(inspect.getdoc(my_func))  # requires an Args:/Arguments:/Parameters: header

concurrency.py

import weakref
from autourgos_core import warn_once_per_agent

_warned_agents: "weakref.WeakSet" = weakref.WeakSet()
warn_once_per_agent(_warned_agents, agent, logger, "this only fires once per agent")

from autourgos_core import retry_with_backoff

result = retry_with_backoff(
    lambda: call_flaky_api(),
    max_attempts=3, backoff_base=1.0, max_backoff=30.0,
    should_retry=lambda exc: not isinstance(exc, PermissionError),
    on_retry=lambda exc, attempt, delay: logger.info(f"retry {attempt} in {delay}s: {exc}"),
)

from autourgos_core import PerAgentRegistry

# for middleware shared across multiple agents -- keeps each agent's state
# isolated, dropped automatically when that agent is garbage collected
_runs: "PerAgentRegistry[dict]" = PerAgentRegistry()
_runs.set(agent, {"exposed": set()})
run_state = _runs.get(agent, default_factory=dict)  # creates+stores if missing
_runs.peek(agent)   # read without creating
_runs.pop(agent, None)

from autourgos_core import RunScopedState

# for middleware whose state must survive being offloaded across a
# hook-executor thread pool WITHIN one invoke()/ainvoke() call, without
# leaking into a concurrent run (two threads, or two interleaved asyncio
# tasks on the same event-loop thread)
_run_ctx: "RunScopedState[dict]" = RunScopedState(default_factory=dict)
_run_ctx.get()["query"] = "..."   # creates the per-run dict on first access
_run_ctx.reset()                  # start a fresh run in the current context

sqlite_utils.py

from autourgos_core import open_sqlite, row_cap_evict

conn = open_sqlite(db_path)  # creates db_path's parent dir, sets PRAGMA journal_mode=WAL
conn.execute("CREATE TABLE IF NOT EXISTS items (id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT)")
conn.execute("INSERT INTO items (value) VALUES (?)", ("x",))
row_cap_evict(conn, "items", "id", max_rows=500)  # deletes oldest rows over the cap
conn.commit()

threading_utils.py

from autourgos_core import PendingCallableQueue, LazyBackgroundThread

# post callables from any thread, drain them from the one thread allowed
# to touch the resource (e.g. a GUI toolkit's own event loop)
queue = PendingCallableQueue()
queue.post(lambda: update_widget("hi"))
queue.drain(on_error=lambda exc: logger.exception("callback raised", exc_info=exc))

# lazily start one background thread, block callers until it's ready
def run(handle: LazyBackgroundThread) -> None:
    try:
        setup_resource()
    except Exception as exc:
        handle.mark_failed(exc)
        return
    handle.mark_ready()
    blocking_event_loop()

bg = LazyBackgroundThread(run, timeout=5.0)
bg.ensure_started()  # raises on failure/timeout, no-ops on later calls

versioning.py

from autourgos_core import package_version

__version__ = package_version("autourgos-mypackage", fallback="1.2.3")

Install

pip install -e .[dev]

Test

pytest

License

Apache 2.0