akgentic-core

Zero-dependency actor framework for the Akgentic multi-agent platform. Define agents, exchange typed messages, and compose concurrent workflows — all in-memory with no external services required.

Table of Contents

• Overview
• Installation
• Quick Start
• Architecture
• Messages
• Agents — Akgent
  • Error Handling
• ActorSystem & ActorAddress
• Communication Patterns
• Agent Lifecycle
• State & Configuration
• Orchestrator & Multi-Agent Coordination
• AgentCard — Capability Discovery
• UserProxy — Human-in-the-Loop
• Examples
• Development
• License

Overview

akgentic-core provides the foundational primitives for building actor-based agent systems with zero infrastructure dependencies — no Redis, no HTTP clients, no database drivers. Everything runs in-process.

The package delivers:

• Actor model runtime via Akgent and ActorSystem — isolated agents communicating exclusively through typed messages
• Typed message dispatch via receiveMsg_<Type> convention — no manual routing code
• Actor addressing via ActorAddress — serializable agent references with rich team metadata
• Communication primitives — self.send() for actor-to-actor messaging; tell / ask for external callers via ActorSystem; typed proxy wrappers for method-call syntax over the message bus
• Typed state & config via BaseState / BaseConfig with observer pattern for reactive updates
• Orchestrator — central coordinator for telemetry, team roster, and pub/sub event distribution
• Capability catalog via AgentCard — declarative agent profiles for dynamic discovery
• Human-in-the-loop via UserProxy — bridge between humans and the agent system

  ┌──────────────────────────────────────────────┐
  │                 ActorSystem                  │
  │                                              │
  │  ┌─────────────┐  message  ┌──────────────┐  │
  │  │   AgentA    │ ────────► │    AgentB    │  │
  │  │  (Akgent)   │           │   (Akgent)   │  │
  │  │  state      │ ◄──────── │   state      │  │
  │  └──────┬──────┘  message  └────────┬─────┘  │
  │         │ telemetry       telemetry │        │
  │         └──────────┐    ┌───────────┘        │
  │                 Orchestrator                 │
  │                (team + events)               │
  └──────────────────────────────────────────────┘

Installation

Workspace Installation (Recommended)

This package is designed for use within the Akgentic monorepo workspace:

git clone git@github.com:b12consulting/akgentic-quick-start.git
cd akgentic-quick-start
git submodule update --init --recursive

uv venv
source .venv/bin/activate
uv sync --all-packages --all-extras

All dependencies resolve automatically via workspace configuration.

Standalone Installation

pip install akgentic-core
# or with uv
uv add akgentic-core

Quick Start

Three building blocks are all you need:

from akgentic.core import ActorSystem, Akgent, ActorAddress, BaseConfig, BaseState
from akgentic.core.messages import Message


class GreetMessage(Message):
    text: str


class GreeterAgent(Akgent[BaseConfig, BaseState]):
    def receiveMsg_GreetMessage(self, msg: GreetMessage, sender: ActorAddress) -> None:
        print(f"Hello, {msg.text}!")


system = ActorSystem()

agent = system.createActor(GreeterAgent, config=BaseConfig(name="greeter", role="Greeter"))
system.tell(agent, GreetMessage(text="Akgentic"))

system.shutdown()

Output:

Hello, Akgentic!

Architecture

akgentic-core wraps the Pykka actor runtime behind a framework-aware abstraction layer. Application code must never use Pykka directly — all interaction goes through Akgent, ActorSystem, and ActorAddress.

┌──────────────────────────────────────────────────────────┐
│  Application Layer: Akgent subclasses, message handlers  │
├──────────────────────────────────────────────────────────┤
│  Framework Layer: ActorSystem, Orchestrator, AgentCard   │
│                   ActorAddress, BaseState, BaseConfig    │
├──────────────────────────────────────────────────────────┤
│  Runtime Layer: Pykka (ThreadingActor, ActorRegistry)    │
└──────────────────────────────────────────────────────────┘

Package Structure

src/akgentic/core/
    __init__.py             # Public API — flat imports
    agent.py                # Akgent base class, ProxyWrapper
    actor_system_impl.py    # ActorSystem, ExecutionContext, Statistics
    actor_address.py        # ActorAddress ABC
    actor_address_impl.py   # ActorAddressImpl, ActorAddressProxy, ActorAddressStopped
    agent_card.py           # AgentCard — capability profiles
    agent_config.py         # BaseConfig, AgentConfig alias
    agent_state.py          # BaseState with observer pattern
    orchestrator.py         # Orchestrator, EventSubscriber, Timer
    user_proxy.py           # UserProxy — human-in-the-loop bridge
    messages/
        message.py          # Message, UserMessage, ResultMessage, StopRecursively
        orchestrator.py     # Telemetry messages (SentMessage, ErrorMessage, …)
    utils/
        serializer.py       # SerializableBaseModel (internal)
        deserializer.py     # ActorAddressDict, DeserializeContext (internal)
examples/                   # 6 progressive examples with companion docs
tests/

Why Pykka Is Abstracted

Pykka is a general-purpose actor library with no awareness of agents, teams, or workflows. The abstraction adds what the framework needs:

Pykka primitive	Framework equivalent	What is added
ThreadingActor	Akgent	Message dispatch, state, telemetry, child creation
ActorRef	ActorAddress	Team metadata, serialization, typed proxy access
ActorRegistry + start()	ActorSystem.createActor()	team_id propagation, orchestrator wiring

Messages

A Message is the only way agents interact. Define message types by subclassing Message:

from akgentic.core.messages import Message

class TaskMessage(Message):
    task_id: str
    payload: str

Every message automatically carries:

• id — unique UUID
• timestamp — creation time
• sender / recipient — ActorAddress references
• team_id — team scope
• parent_id — causal chain tracking

Messages are immutable data packets. Import business messages from akgentic.core.messages:

from akgentic.core.messages import (
    Message,           # Base class for all application messages
    UserMessage,       # Human input into the agent system
    ResultMessage,     # Agent response to a UserMessage
    StopRecursively,   # Signal recursive shutdown
)

Telemetry messages (SentMessage, ReceivedMessage, ErrorMessage, etc.) flow automatically to the Orchestrator. Import them when building EventSubscriber implementations or handling errors programmatically:

from akgentic.core.messages.orchestrator import (
    SentMessage, ReceivedMessage, ProcessedMessage, ErrorMessage,
    StartMessage, StopMessage, StateChangedMessage, EventMessage,
)

Agents — Akgent

Akgent[ConfigType, StateType] is the base class every agent extends. It turns a raw Pykka actor into a framework agent:

from akgentic.core import Akgent, BaseConfig, BaseState, ActorAddress

class SummaryAgent(Akgent[BaseConfig, BaseState]):

    def on_start(self) -> None:
        """Initialisation hook — runs inside the actor thread after startup."""
        self.state = BaseState()
        self.state.observer(self)

    def receiveMsg_TaskMessage(self, msg: TaskMessage, sender: ActorAddress) -> None:
        """Handler name = receiveMsg_ + message class name."""
        result = self._summarize(msg.payload)
        self.send(sender, ResultMessage(content=result))

    def _summarize(self, text: str) -> str:
        return text[:100]

Key conventions:

• receiveMsg_<ClassName> — automatic dispatch; no manual routing needed
• on_start() — always initialise state here, never in __init__
• self.send(recipient, message) — send from within an actor
• self.myAddress — obtain own ActorAddress for self-reference

Key methods:

Method	Description
on_start()	Initialisation hook (actor thread)
send(recipient, msg)	Send message with telemetry
createActor(cls, config)	Spawn child actor with context propagation
stop()	Recursive stop (children first, then self)
update_state(updates)	Merge dict into typed state
notify_event(event)	Emit domain event via EventMessage
proxy_tell(addr, Type)	Typed fire-and-forget proxy call
proxy_ask(addr, Type)	Typed blocking proxy call
get_team()	Team roster via orchestrator
get_agent_card(role)	Look up capability profile
find_agents_with_skill(skill)	Discover agents by skill

Error Handling

When an unhandled exception occurs during message processing, Akgent uses Pykka's _handle_failure() hook (not a try/except wrapper around dispatch):

1. Log the error with full context
2. Emit ProcessedMessage to the orchestrator (marks the current message as done)
3. Check for WarningError — if so, silently acknowledge and return
4. Emit ErrorMessage with exception_type, exception_value, traceback, and current_message to the orchestrator

The actor does not crash — it continues processing subsequent messages.

WarningError is a soft signal for non-critical failures (e.g., usage limits exceeded). Raise it from a message handler when the error should be logged and the current message marked as processed, but no ErrorMessage should be sent to the orchestrator. Import it from akgentic.core:

from akgentic.core import WarningError

class MyAgent(Akgent[BaseConfig, BaseState]):
    def receiveMsg_TaskMessage(self, msg: TaskMessage, sender: ActorAddress) -> None:
        if self._over_budget():
            raise WarningError("Usage limit exceeded")  # logged, no ErrorMessage

For proxy ask() calls, Pykka's reply mechanism handles errors automatically — the exception is sent back to the caller, bypassing _handle_failure().

ActorSystem & ActorAddress

ActorSystem

ActorSystem is the sole gateway between external code and the actor world. From outside an actor (a web handler, a test, a CLI), all interaction goes through ActorSystem:

system = ActorSystem()

# Spawn an agent — returns an ActorAddress, never a direct object reference
agent = system.createActor(MyAgent, config=BaseConfig(name="agent", role="MyAgent"))

# Fire-and-forget
system.tell(agent, MyMessage(data="hello"))

# Blocking request — wait for handler's return value
result = system.ask(agent, QueryMessage(query="..."), timeout=10.0)

# Receive a reply sent back to the system context
response = system.listen(timeout=5.0)

# Typed proxy — method call syntax, still message-passing under the hood
proxy = system.proxy_ask(agent, MyAgent, timeout=5.0)
result = proxy.some_method(arg)

system.shutdown()

Use system.private() when you need an isolated context for scripted workflows or integration tests where the caller receives replies directly:

with system.private() as ctx:
    ctx.tell(agent, MyMessage())
    reply = ctx.listen(timeout=5.0)

ActorAddress

ActorAddress is a reference to an agent — like a mailbox address. You never hold a direct Python object reference to another agent.

addr.agent_id   # UUID — unique agent identity
addr.name       # str  — e.g. "@Summarizer"
addr.role       # str  — e.g. "SummaryAgent"
addr.team_id    # UUID — always set; defines team membership
addr.is_alive() # bool — whether the actor is still running
addr.serialize()# → ActorAddressDict — survives serialization/persistence

Three implementations cover the full actor lifecycle:

Class	Used when	send()
ActorAddressImpl	Live actor	delivers to mailbox
ActorAddressProxy	Deserialized / mock	raises RuntimeError
ActorAddressStopped	Post-stop tracking	raises RuntimeError

Communication Patterns

tell vs ask

	tell / proxy_tell	ask / proxy_ask
Blocks caller	No — fire-and-forget	Yes — until handler returns
Return value	None	Handler's return value
Deadlock risk	None	Yes if called from within the same actor
Use for	Notifications, events	Queries, request-response

Bidirectional Messaging (reply via sender)

Every receiveMsg_<Type> handler receives sender: ActorAddress. Reply by sending a message back:

class ResponderAgent(Akgent[BaseConfig, BaseState]):
    def receiveMsg_QueryMessage(self, msg: QueryMessage, sender: ActorAddress) -> None:
        result = self._compute(msg.query)
        self.send(sender, ResultMessage(content=result))

Typed Proxy Wrappers

proxy_tell and proxy_ask provide method-call syntax over the message bus — the actor model principle is preserved because every call is still converted to a mailbox message internally:

# Outside the actor system
orch_proxy = system.proxy_ask(orchestrator_addr, Orchestrator)
team = orch_proxy.get_team()           # → ask() → mailbox → handler → return

# Inside an actor (actor-to-actor)
worker_proxy = self.proxy_tell(worker_addr, WorkerAgent)
worker_proxy.process(task)             # → tell() → worker's mailbox

Agent Lifecycle

Spawning Agents

Agents are created with createActor() — either from ActorSystem (root actors) or from within an actor (child actors):

# Root actor — from outside
orchestrator = system.createActor(
    Orchestrator,
    config=BaseConfig(name="orchestrator", role="Orchestrator"),
)

# Child actor — from inside an agent
class ManagerAgent(Akgent[BaseConfig, BaseState]):
    def on_start(self) -> None:
        self._worker = self.createActor(
            WorkerAgent,
            config=WorkerConfig(name="worker-1"),
        )
        # team_id and orchestrator reference are automatically propagated

When spawning through a parent, three things propagate automatically:

• team_id — child joins the same team
• orchestrator — child reports telemetry to the same coordinator
• parent — stored as self._parent on the child

on_start() Hook

Always perform actor initialisation in on_start(), never in __init__. on_start() runs inside the actor thread after startup, making it safe to create child actors and attach state observers:

class MyAgent(Akgent[MyConfig, MyState]):
    def on_start(self) -> None:
        self.state = MyState()
        self.state.observer(self)          # reactive state updates
        self._child = self.createActor(HelperAgent)

Stopping

stop() cascades recursively — children are stopped before the parent. To shut down a team, stop the Orchestrator:

orchestrator.stop()
  → stops team members (recursively)
  → stops orchestrator itself
  → sends StopMessage to telemetry log

State & Configuration

BaseConfig

BaseConfig is the typed configuration model for an agent. Subclass it to add agent-specific fields:

from akgentic.core import BaseConfig

class WorkerConfig(BaseConfig):
    max_retries: int = 3
    timeout: float = 30.0

Configuration is injected at creation and accessible as self.config throughout the agent's lifetime. When agents are instantiated from an AgentCard, get_config_copy() returns a deep copy — preventing shared mutable state across instances.

BaseState

BaseState is a Pydantic model with an observer pattern. State changes automatically notify the Orchestrator via StateChangedMessage:

from akgentic.core import BaseState

class WorkerState(BaseState):
    tasks_completed: int = 0
    current_task: str | None = None

class WorkerAgent(Akgent[WorkerConfig, WorkerState]):
    def on_start(self) -> None:
        self.state = WorkerState()
        self.state.observer(self)          # attach — triggers initial notification

    def receiveMsg_TaskMessage(self, msg: TaskMessage, sender: ActorAddress) -> None:
        self.update_state({
            "current_task": msg.task_id,
            "tasks_completed": self.state.tasks_completed + 1,
        })
        # Orchestrator is notified automatically

update_state(self, updates: dict[str, Any]) -> None performs a full Pydantic round-trip: merges updates into model_dump(), deserializes via AkgentDeserializeContext, then calls init_state() which preserves the observer and notifies.

Note — direct field mutation does not auto-notify. update_state() is the only path that notifies the Orchestrator automatically. If you mutate a field on self.state directly instead (e.g. self.state.count += 1), Pydantic attribute assignment does not trigger any hook — you must call self.state.notify_state_change() yourself afterward, or the Orchestrator's state_dict and any EventSubscribers never learn about the change:

self.state.count += effective
self.state.notify_state_change()   # required after direct mutation

Orchestrator & Multi-Agent Coordination

The Orchestrator

The Orchestrator is always the root actor of a team. It serves as the central coordinator for:

• Telemetry — records every lifecycle event and message exchange (including EventMessage)
• Team roster — tracks which agents are alive via StartMessage/StopMessage
• State snapshots — stores the latest BaseState for each agent
• Pub/sub — distributes events to EventSubscriber implementations

from akgentic.core import Orchestrator, BaseConfig

orchestrator_addr = system.createActor(
    Orchestrator,
    config=BaseConfig(name="orchestrator", role="Orchestrator"),
)
# team_id is generated here — this becomes the team's identity

# Spawn all other agents through the orchestrator so they inherit team_id
agent_addr = orchestrator_addr.createActor(MyAgent, ...)

Team management (via proxy):

orch = system.proxy_ask(orchestrator_addr, Orchestrator)

orch.get_team()                    # Active agent addresses (excludes Orchestrator)
orch.get_team_member("@Writer")    # Find by name
orch.get_messages()                # Full telemetry log
orch.get_states()                  # Latest state per agent
orch.get_events()                  # All EventMessages (optional agent_id/event_class filters)
orch.get_metadata()                # Team-scoped business context (None if unset)
orch.set_metadata(metadata)        # Replace it wholesale (None clears it)

team_id Inheritance

All non-orchestrator agents must be spawned through the Orchestrator (or through an agent already in the team). Direct creation from ActorSystem gives an isolated team_id — the agent will not appear in get_team() and its telemetry will not flow to the Orchestrator.

ActorSystem.createActor(Orchestrator)   → team_id = <UUID-A>
  └─ Orchestrator.createActor(AgentA)   → team_id = <UUID-A>  (propagated)
       └─ AgentA.createActor(AgentB)    → team_id = <UUID-A>  (propagated again)

Event Subscribers

Subscribe to the telemetry stream for persistence, streaming, or external integrations:

from akgentic.core import EventSubscriber
from akgentic.core.messages import Message

class MySubscriber(EventSubscriber):
    def on_message(self, msg: Message) -> None:
        print(f"[telemetry] {type(msg).__name__}")

    def on_stop(self) -> None:
        pass

orch.subscribe(MySubscriber())

on_message() receives all telemetry types: StartMessage, StopMessage, SentMessage, ReceivedMessage, ProcessedMessage, ErrorMessage, StateChangedMessage, EventMessage.

Team Metadata

team_metadata is caller-defined, team-scoped business context — tenant, case reference, channel, department — that any agent in the team can read at runtime through the Orchestrator. It is opaque to core: the value arrives as an already-validated SerializableBaseModel subclass, and core stores and returns it unchanged, never validating, inspecting, or indexing it. The schema and the filtering built on it live in akgentic-team.

from akgentic.core import Orchestrator
from akgentic.core.utils import SerializableBaseModel

class CaseContext(SerializableBaseModel):
    tenant: str
    case: str

orch = system.proxy_ask(orchestrator_addr, Orchestrator)

orch.set_metadata(CaseContext(tenant="acme", case="C-1234"))
ctx = orch.get_metadata()           # CaseContext(tenant='acme', case='C-1234')
orch.set_metadata(None)             # clears it

set_metadata(metadata) replaces the value wholesale — it never merges, so what is set does not depend on write history. get_metadata() returns the caller's own subclass by reference; treat it as read-only and call set_metadata() with a new model to change it.

Setting the value emits no StateChangedMessage. The value is therefore not part of any agent state snapshot, and an EventSubscriber will not observe metadata writes on the telemetry stream — a snapshot would become a second persisted copy, free to diverge from the record that team listing indexes.

Note — the Orchestrator's copy is a cache, not the system of record. The authoritative value lives in akgentic-team's Process record. The team layer writes that record first and only then pushes to the live actor, best-effort, so after a failed push the actor's copy can legitimately lag until the next team resume repopulates it. Code that needs the authoritative value must read Process, not get_metadata(). This is also the one value the telemetry replay described under Team Restoration below does not bring back — no metadata write ever reaches the telemetry log, so akgentic-team repopulates it from Process as part of restoring the team.

Team Restoration

The Orchestrator's telemetry log is the single source of truth for crash recovery. Because every lifecycle and business event flows through it, a team can be fully reconstructed by:

1. Identifying agents alive at shutdown (StartMessage minus StopMessage)
2. Recreating those actors with original agent_id, team_id, and config
3. Replaying persisted events via restore_message() to rebuild in-memory state

akgentic-team implements the full 3-phase restore protocol on top of these primitives. See akgentic-team for details.

AgentCard — Capability Discovery

AgentCard is a declarative profile that describes an agent type. Register profiles with the Orchestrator so running agents can discover capabilities without hardcoding dependencies:

from akgentic.core import AgentCard, BaseConfig

card = AgentCard(
    role="ResearchAgent",
    description="Performs web research and data gathering",
    skills=["web_search", "pdf_extraction"],
    agent_class=ResearchAgent,             # class or fully-qualified string
    config=BaseConfig(name="researcher", role="ResearchAgent"),
    routes_to=["WriterAgent"],             # empty = no routing restrictions
)

# Register with the Orchestrator
orch.register_agent_profile(card)

# Query the catalog
orch.get_agent_catalog()                   # all profiles
orch.get_agent_profile("ResearchAgent")    # by role
orch.get_profiles_by_skill("web_search")   # by skill
orch.get_available_roles()                 # role list

From within an agent, use the built-in discovery methods:

class CoordinatorAgent(Akgent[BaseConfig, BaseState]):
    def receiveMsg_PlanMessage(self, msg, sender):
        writers = self.find_agents_with_skill("writing")
        card = self.get_agent_card("ResearchAgent")
        config = card.get_config_copy()    # deep copy — safe to mutate

Profile vs. instance:

AgentCard catalog  → "What agent types exist?" (static capability directory)
get_team()         → "What instances are running?" (dynamic runtime roster)

routes_to routing constraints:

• Empty list → no restrictions; the agent can send to any role
• Non-empty list → restricted; only listed roles are valid targets
• Responses are always allowed regardless of routes_to

UserProxy — Human-in-the-Loop

UserProxy is a regular team actor that acts as the boundary between the agent system and a human user. The interaction follows a two-leg flow:

Agent ──UserMessage──►  UserProxy  ──(telemetry)──►  Orchestrator
                                                           │
                                              EventSubscriber (e.g. WebSocket)
                                                           │
                                                      external UI
                                                           │
                                        ActorSystem.proxy_ask(user_proxy_addr, UserProxy)
                                                           │
                                    Agent ◄── process_human_input(content, msg)

Leg 1 — forwarding to the human: When an agent needs human input it sends a UserMessage to the UserProxy actor. receiveMsg_UserMessage fires in the proxy's thread. The default implementation only logs — the message flows through the Orchestrator as normal telemetry, so any registered EventSubscriber can intercept it and forward it to the external system.

Leg 2 — injecting the human's response: When the human replies, the external system calls process_human_input() on the UserProxy via an ActorSystem proxy call. The default implementation wraps the response in a ResultMessage and sends it back to msg.sender — the agent that originally asked.

from akgentic.core import UserProxy, UserMessage, ActorAddress

# Subclass to integrate with your UI
class MyUserProxy(UserProxy):
    def receiveMsg_UserMessage(self, msg: UserMessage, sender: ActorAddress) -> None:
        # log the message in the Orchestrator telemetry (received/processed messages)
        pass

# Spawn via the Orchestrator like any other team member
proxy_addr = orchestrator_addr.createActor(
    MyUserProxy,
    config=BaseConfig(name="@Human", role="UserProxy"),
)

# When the human replies, the external system injects the answer.
# Pass the original UserMessage so the proxy knows who to reply to.
proxy = system.proxy_ask(proxy_addr, MyUserProxy)
proxy.process_human_input("Approved", original_user_message)  # original_user_message: the UserMessage received in Leg 1

akgentic-agent provides HumanProxy, a richer subclass that handles multi-hop routing via continuation chains — useful when the request travels through several agents before reaching the human (e.g. Manager → Dev → Human → Dev → Manager). See akgentic-agent for details.

Examples

Six progressive, self-contained examples in the examples/ directory. Each includes a runnable .py script and a companion .md explaining concepts and pitfalls.

uv run python examples/01_hello_world.py

#	Script	Topic
01	01_hello_world.py	Message, Akgent, ActorSystem — first agent
02	02_request_response.py	Bidirectional messaging, tell vs ask, proxy wrappers
03	03_dynamic_agents.py	createActor(), parent-child hierarchy, on_start()
04	04_stateful_agents.py	BaseConfig, BaseState, observer pattern, Orchestrator
05	05_multi_agent.py	Multi-agent workflows, UserProxy, EventSubscriber
06	06_agent_cards.py	AgentCard, capability catalog, routing constraints

See examples/README.md for the full concept index.

Development

Prerequisites

• Python 3.12+
• uv package manager

Setup

uv sync --all-extras

Commands

All commands run from the monorepo root (akgentic-quick-start/):

# Run tests
pytest packages/akgentic-core/tests/

# Run tests with coverage
pytest packages/akgentic-core/tests/ --cov=akgentic.core --cov-fail-under=80

# Lint
ruff check packages/akgentic-core/src/

# Format
ruff format packages/akgentic-core/src/

# Type check
mypy packages/akgentic-core/src/

License

This project is licensed under the GNU Affero General Public License v3.0 (AGPL-3.0).

Dual licensing & CLA — Akgentic is available under the AGPL-3.0 open-source license. A commercial license is also planned for organizations that require alternative terms. Contact Yuma for more information. External contributions will be accepted once a Contributor License Agreement (CLA) is in place. Until then, please hold off on submitting pull requests.