Start building agents with akgents ~ the core

Why I started with the core

#

Agent frameworks often make a great first impression. Give an agent a model, a prompt, and a few tools, and something interesting happens. The difficult questions arrive one step later. Where does the changing state live? How do agents find each other? What happens when a human needs to approve a change? And how can I see what a team did after it finishes?

I want to understand how Akgents answers those questions. The framework contains more than its core, but I deliberately start this series with the akgentic-core package. Before adding models, tools, memory, or other capabilities, I want a solid mental model of agents, messages, state, orchestration, and lifecycle.

To make those concepts concrete, I built a small case-triage application. It does not call an LLM. That is a feature for this first step: every interaction you see comes from the framework and the application design, so there is no model behaviour hiding what the actors are doing.

Akgents and the actor model

#

Akgents is a Python framework for building multi-agent systems. Its packages separate the actor foundation from optional capabilities such as LLM integration, tools, team composition, catalogs, infrastructure, and a frontend. In this article I stay with akgentic-core: the layer that provides actors, typed messages, and orchestration without requiring those higher-level capabilities.

Under that core sits Pykka’s implementation of the actor model. An actor is an independently executing unit that owns its state instead of sharing it directly. It communicates by sending messages to actors whose addresses it knows. After receiving a message, an actor can update its own state, send more messages, or create another actor.

The rule that matters most for this sample is that one actor processes one message at a time. That makes an actor the natural owner of mutable data: its handlers can change its state without an internal lock, while other actors interact with it through messages. This is why the case repository is represented by CaseRepositoryAgent; instead of sharing a repository across threads, the team gives it one owner and sends requests for every read and update.

The case-triage sample

#

The application creates one lightweight team for one case. A coordinator accepts the request, a triage agent inspects the case, and a user proxy brings the proposed priority to a human. The team does the work; the human makes the decision.

The sample currently contains four agents:

• CaseCoordinatorAgent coordinates the complete flow.
• CaseTriageAgent assesses the case and proposes its priority.
• CaseRepositoryAgent owns the case store and handles every read and write.
• CliUserProxyAgent connects the actor team to a person using the command line.

Behind CaseRepositoryAgent is a CaseRepository interface with an in-memory implementation. It represents the case system without coupling the team to a database or API. Only the repository agent holds this implementation; the other agents request case data and updates with messages.

The wider case-management idea can grow into agents for gathering information and handling different types of cases. I leave those out of this episode. Four agents are enough to expose the core patterns without turning the sample into a framework tour.

Follow the sample alongside the series

#

The basic-akgents sample repository is the companion to this series. The blog_part_1 branch contains the version used in this episode, so you can keep the code open while reading and run each example yourself.

I create a separate branch for every episode. That keeps the code for this article stable while the sample continues to evolve. By moving through the episode branches, you can follow both the blogs and the progression of the application step by step.

Run the sample

#

The project uses Python 3.12 or newer and uv for dependency management. The dependency is small:

[project]
requires-python = ">=3.12"
dependencies = [
    "akgentic-core>=1.5.4",
]

When starting a new project, add the core with:

uv add akgentic-core

For a checkout of this sample, restore the locked environment and run a case:

uv sync
uv run src/main.py case_2

case_2 describes a mail server outage. Triage recognises the word urgent, proposes the critical priority, and includes its reason in the question:

[Case case_2] Team started. Triage assesses the case, you approve its priority.

[@CaseCoordinator] Case case_2
  description : Mail server is down for the whole department, urgent
  reported by : jettrocoenradie
  proposal    : priority 1 - critical, because the case mentions 'urgent'
  scale       : 1 - critical | 2 - high | 3 - normal | 4 - low (1 is the most urgent)
Approve priority 1 - critical? [Enter/y] approve, [n] reject, or type another number (1-4)
>

Pressing Enter approves the proposal. Entering n rejects it, while a number from 1 through 4 overrides the proposal and approves the chosen priority in one step.

The dummy repository contains several scenarios that reveal different paths through the team:

The last two paths complete with 22 recorded messages instead of the 40 used by the approval path. That difference becomes visible through the orchestrator telemetry.

The actor system, orchestrator, and team

#

The first Akgents concept that clicked for me is that the orchestrator is not an invisible service created by the actor system. It is an Akgent itself, and the application creates it explicitly.

flowchart TD
    AS["ActorSystem"] --> OR["@Orchestrator"]
    OR --> CO["@CaseCoordinator"]
    CO --> TR["@CaseTriage"]
    CO --> RE["@CaseRepository"]
    CO --> UP["@UserProxy"]

The distinction affects how the application starts. ActorSystem.createActor starts the orchestrator. The orchestrator then creates the coordinator, and the coordinator creates the other team members:

actor_system = ActorSystem()

orchestrator_addr = actor_system.createActor(
    Orchestrator,
    config=BaseConfig(name="@Orchestrator", role="Orchestrator"),
)
orchestrator = actor_system.proxy_ask(orchestrator_addr, Orchestrator)

coordinator_addr = orchestrator.createActor(
    CaseCoordinatorAgent,
    config=CaseCoordinatorConfig(
        name="@CaseCoordinator",
        role="Coordinator",
        case_id=case_id,
    ),
)
coordinator = actor_system.proxy_ask(coordinator_addr, CaseCoordinatorAgent)

triage_agent_address = coordinator.createActor(
    CaseTriageAgent,
    config=CaseTriageConfig(
        name="@CaseTriage",
        role="Triage",
        case_id=case_id,
    ),
)

repository_agent_address = coordinator.createActor(
    CaseRepositoryAgent,
    config=CaseRepositoryConfig(
        name="@CaseRepository",
        role="Repository",
        case_id=case_id,
    ),
)

Creating agents through this hierarchy propagates the orchestrator address, team id, and parent. If I create the coordinator directly on the actor system, it runs, but it has no orchestrator. Team lookup, state snapshots, and message telemetry then disappear. The hierarchy is therefore part of the runtime model, not just an organisational diagram.

What makes an Akgent

#

An Akgent combines four elements:

1. A configuration model for facts that identify or configure the agent.
2. A state model for information that changes while messages are handled.
3. Message classes that describe what agents exchange.
4. Handlers named receiveMsg_<MessageType>.

The type parameters make the configuration and state explicit:

class CaseTriageAgent(Akgent[CaseTriageConfig, CaseTriageState]):
    def on_start(self) -> None:
        self.state = CaseTriageState()
        self.repository_agent = None
        self.reply_to = None
        self.state.observer(self)

    def receiveMsg_CaseTriageRequest(
        self,
        message: CaseTriageRequest,
        sender: ActorAddress,
    ) -> None:
        self.reply_to = sender
        self.update_state({
            "status": "loading",
            "requester_id": message.requester_id or "unknown",
        })
        self._ask_repository(
            CaseInformationRequest(case_id=self.config.case_id)
        )

on_start runs in the actor’s own thread. This is where the sample creates the concrete state, initialises live instance fields, and connects the state observer. Each agent processes its own mailbox sequentially, so an agent does not need a lock around its state.

The messages are Pydantic models derived from Message. They make the conversation visible and serialisable:

class CaseTriageResponse(Message):
    case_description: str = ""
    case_sender: str = ""
    case_priority: CasePriority = CasePriority.UNSET
    reason: str = ""
    known_case: bool = False
    already_prioritised: bool = False

This response communicates more than a priority. It tells the coordinator whether the case exists, whether triage still needs to act, and why it proposed a value. The coordinator can react without reaching into the triage agent.

Config is identity; state is progress

#

The practical rule I use is simple:

Config contains facts that stay fixed for the lifetime of an agent. State contains everything that changes while the agent handles messages.

Every agent in the team works on the same case, so the case id is part of its identity:

class CaseConfig(BaseConfig):
    case_id: str = ReadOnlyField(frozen=True)


class CaseCoordinatorConfig(CaseConfig):
    pass

ReadOnlyField marks the field as read-only in the JSON schema. frozen=True is the part that stops assignment at runtime. I freeze this field, not the complete config model, because the framework can fill the inherited name and role fields during startup.

The coordinator state contains the moving parts of the workflow:

class CaseCoordinatorState(BaseState):
    status: str = "new"
    requester_id: str = ""
    case_description: str = ""
    proposed_priority: CasePriority = CasePriority.UNSET

When the coordinator calls update_state, Akgents replaces the state model and notifies the orchestrator. That produces the snapshots shown in the final summary. The case_id does not belong there: repeating an invariant in every state snapshot adds noise and creates two possible sources of truth.

The repository-agent design makes state important in two additional places. Triage moves through statuses such as loading, proposed, and storing, so it knows which response it is waiting for without blocking its thread. CaseRepositoryState counts reads and writes and records the last case id, making access to the store visible through the orchestrator.

Both config and state become part of the event and telemetry stream. That makes them useful for observability and reactivation, but it also leads to an important rule: never store secrets in either one.

Connect agents with messages and addresses

#

An ActorAddress is live runtime wiring. It belongs on the agent instance, not in serialised state. The coordinator prepares the slots in on_start, and the composition root wires the addresses after creating the agents:

class CaseCoordinatorAgent(Akgent[CaseCoordinatorConfig, CaseCoordinatorState]):
    def on_start(self) -> None:
        self.state = CaseCoordinatorState()
        self.triage_agent = None
        self.user_proxy = None
        self.state.observer(self)

    def set_agents(
        self,
        triage_agent_address: ActorAddress,
        user_proxy_address: ActorAddress,
    ) -> None:
        self.triage_agent = triage_agent_address
        self.user_proxy = user_proxy_address

Once connected, agents collaborate with send:

self.send(
    self.triage_agent,
    CaseTriageRequest(requester_id=message.requester_id),
)

This is the normal path. A sent message carries its sender, recipient, team, and parent relationship, which preserves the conversation and generates telemetry. Direct proxy calls are useful for wiring local objects, but using them for the business conversation would make the flow harder to observe.

Triage also receives the address of the repository agent. It stores that live address on the instance and uses it for case-information and update messages:

actor_system.proxy_tell(
    triage_agent_address,
    CaseTriageAgent,
).set_repository_agent(repository_agent_address)

Give shared state a single owner

#

The team needs case data, but it should not know whether that data comes from memory, a database, or a remote service. The sample still expresses that backend boundary as a Python Protocol:

@runtime_checkable
class CaseRepository(Protocol):
    def load_case(self, case_id: str) -> Case: ...
    def save_case(self, case: Case) -> None: ...

DummyCaseRepository satisfies the protocol without inheriting from it. The composition root chooses the implementation, but only CaseRepositoryAgent receives the live object:

case_repository: CaseRepository = DummyCaseRepository()

actor_system.proxy_tell(
    repository_agent_address,
    CaseRepositoryAgent,
).set_case_repository(case_repository)

A repository is a live object. It cannot go into config or state because both are serialised, and createActor does not accept arbitrary constructor arguments. The setter keeps the backend on the repository-agent instance. Every other team member only knows its ActorAddress.

This gives the mutable case data one owner. Akgents processes one agent’s mailbox sequentially, so CaseRepositoryAgent handles only one repository operation at a time. The in-memory repository no longer needs a lock because no other agent touches it. More importantly, the repository agent performs load, change, and save inside one message handler. Another writer cannot slip between those steps and overwrite part of the audit log.

The messages describe intent rather than exposing repository methods:

class CaseInformationRequest(Message):
    case_id: str = ""


class CaseUpdateRequest(Message):
    case_id: str = ""
    case_priority: CasePriority = CasePriority.UNSET
    action: str = ""

CaseInformationResponse returns the current Case. CaseUpdateRequest asks the repository agent to apply a priority and append an audit action. UNSET means that the current priority must remain untouched, which is exactly what a rejected proposal needs.

The trade-off is explicit asynchronous flow. Triage can no longer call load_case and immediately use its return value. It sends a request, records status="loading", and continues when CaseInformationResponse arrives. Storing follows the same pattern:

CaseTriageRequest    -> CaseInformationRequest -> CaseInformationResponse -> CaseTriageResponse
CasePriorityDecision -> CaseUpdateRequest      -> CaseUpdateResponse      -> CaseTriageCompleted

Response handlers check the status before continuing. This is the same actor principle used for human approval: an actor waits in its state, never by blocking its thread. The extra messages also make every repository access observable. CaseRepositoryState exposes its read and write counters in the orchestrator snapshots.

One boundary remains important: the no-lock guarantee depends on the repository agent being the only owner of that repository instance. A blocking database or network call will also block the repository agent’s mailbox, so a production implementation still needs appropriate I/O behaviour.

Keep the human in the loop without blocking the workflow

#

The most useful part of the sample is not that it asks a person a question. It is how responsibility is divided. Triage reads the case and proposes a priority with a reason. The human approves, rejects, or changes that proposal. The person decides; the agent team does the work.

sequenceDiagram
    participant Main as "main"
    participant Coordinator as "@CaseCoordinator"
    participant Triage as "@CaseTriage"
    participant Repository as "@CaseRepository"
    participant Proxy as "@UserProxy"
    participant Human as "Human"

    Main->>Coordinator: HandleCaseRequest
    Coordinator->>Triage: CaseTriageRequest
    Triage->>Repository: CaseInformationRequest
    Repository-->>Triage: CaseInformationResponse
    Triage-->>Coordinator: CaseTriageResponse + reason
    Coordinator->>Proxy: UserMessage
    Proxy->>Human: Ask for approval
    Human-->>Proxy: Approve, reject, or override
    Proxy-->>Coordinator: ResultMessage
    Coordinator->>Triage: CasePriorityDecision
    Triage->>Repository: CaseUpdateRequest
    Repository-->>Triage: CaseUpdateResponse
    Triage-->>Coordinator: CaseTriageCompleted
    Coordinator->>Proxy: ResultMessage
    Proxy-->>Main: completion event

The coordinator models the open question in state:

self.update_state({
    "status": "awaiting_approval",
    "case_description": message.case_description,
    "proposed_priority": message.case_priority,
})

self.send(self.user_proxy, UserMessage(content=question))

It then returns and remains able to process messages. When the proxy sends a ResultMessage, the coordinator only accepts it while the state is awaiting_approval. It translates the answer into a CasePriorityDecision and sends that decision to triage.

This separation also controls side effects. The information response lets triage compute a proposal but never save it. After the human answers, triage translates CasePriorityDecision into a CaseUpdateRequest. Only the repository agent applies that intent. Rejection leaves the priority unset and adds an audit entry; approval stores the selected priority and who approved it.

The CLI proxy itself does block on input(), which is acceptable for this demonstration. It blocks only the proxy’s mailbox, not the coordinator. A web or event-driven user interface would replace that input mechanism while preserving the same UserMessage and ResultMessage exchange.

Make domain rules explicit

#

Agent code still needs an unambiguous domain model. A plain integer priority forces every reader to guess whether 1 or 4 is more urgent. The sample makes the scale explicit:

class CasePriority(IntEnum):
    UNSET = 0
    CRITICAL = 1
    HIGH = 2
    NORMAL = 3
    LOW = 4

IntEnum keeps the values sortable and serialisable as numbers, while the names carry their meaning. The prompt always displays both the value and label. UNSET is not the lowest priority; it means the case has not been triaged.

That distinction makes the rule for existing cases readable:

if case.case_priority.is_set:
    self._reply(
        CaseTriageResponse(
            case_description=case.case_description,
            case_priority=case.case_priority,
            reason=f"the case already has priority {case.case_priority.label}",
            known_case=True,
            already_prioritised=True,
        )
    )
    return

The repository agent owns access to the case data; triage owns the rule that decides whether the case still needs triaging. The coordinator only reacts to the result. An existing priority closes the request without asking a human to approve the same decision again. An unknown case follows the same short route but reports that there was nothing to triage.

Observe and stop the team

#

After the approval flow, the application asks the orchestrator for its view of the team:

team = orchestrator.get_team()
messages = orchestrator.get_messages()
states = orchestrator.get_states()

For case_2, the verified run ends with:

[@CaseCoordinator] Case case_2 has been triaged.
  description : Mail server is down for the whole department, urgent
  priority    : 1 - critical (approved by jettrocoenradie)

=== Orchestrator Summary ===
Total messages: 40
Team members: 4 agents
State snapshots: 4 agents tracked
===========================
[Multi-Agent] Demo complete. Shutting down.

Those 40 messages include the framework’s lifecycle, sent, received, processed, and state-change telemetry—not only the domain messages in the sequence diagram. That record is what makes it possible to inspect the team instead of treating the agents as opaque background threads. The increase from the earlier flow comes from making repository reads and updates observable conversations handled by a fourth agent.

The application also waits for a real completion signal instead of guessing with time.sleep:

case_handled = threading.Event()
actor_system.proxy_tell(
    user_proxy_address,
    CliUserProxyAgent,
).set_completion_event(case_handled)

actor_system.tell(
    coordinator_addr,
    HandleCaseRequest(requester_id=requester_id),
)

case_handled.wait(timeout=300)

The proxy sets the event after receiving the final result. A finally block then calls actor_system.shutdown(timeout=5), ensuring that the actor system is also stopped when something fails. Starting, observing, completing, and shutting down are all part of the sample’s lifecycle.

What to take into your first Akgents project

#

The sample is intentionally small, but it established the foundation I wanted before using more of the framework:

• Create the orchestrator explicitly and build the team through the agent hierarchy.
• Put stable identity and settings in config; put changing progress in state.
• Communicate through messages so the flow remains serialisable and observable.
• Give shared mutable data one agent owner and express reads and updates as messages.
• Keep actor addresses, repositories, and other live objects on the relevant agent instance.
• Let agents prepare work and let humans decide, with no side effects before approval.
• Model waiting as state instead of blocking the coordinator or triage agent.
• Use the orchestrator to inspect the team, messages, and state snapshots.
• Wait for an actual completion signal and shut the actor system down deliberately.

None of these concepts requires an LLM. That is exactly why I wanted to understand them first. When models and tools enter the team in later episodes, they can build on a message flow and lifecycle that are already explicit. The interesting AI behaviour then becomes one part of the system instead of the thing holding the complete system together.