Mozaik TypeScript runtime for concurrent AI agents Mozaik is a TypeScript runtime for non-blocking AI agents that work independently, communicate through semantic events, and coordinate without predefined workflows. Installation npm npm install @mozaik-ai/core yarn yarn add @mozaik-ai/core pnpm pnpm add @mozaik-ai/core API Key Configuration Mozaik picks a provider from the model name you pass to runInference, and each provider's SDK reads its credential from the environment. Set the keys for the providers you use: # .env OPENAI_API_KEY=your-openai-key-here ANTHROPIC_API_KEY=your-anthropic-key-here GEMINI_API_KEY=your-gemini-key-here DeepSeek models run through the OpenAI-compatible chat-completions endpoint, so they use an OpenAI-style credential and base URL (OPENAI_API_KEY / OPENAI_BASE_URL) pointed at DeepSeek. The agentic environment AgenticEnvironment is where everything happens. Participants join() it, and from that moment on they can listen to messages and events flowing through the environment by overriding any of the handlers below: Handler Triggered when… onJoined this participant joins an environment onLeft this participant leaves an environment onParticipantJoined another participant joins the same environment onParticipantLeft another participant leaves the same environment onMessage any participant sends a message onFunctionCall its own inference returns a function call onExternalFunctionCall another agent's inference returns a function call onFunctionCallOutput its own function call runner returns a result onExternalFunctionCallOutput another agent's function call runner returns a result onReasoning its own inference returns a reasoning item onExternalReasoning another agent's inference returns a reasoning item onModelMessage its own inference returns an assistant message onExternalModelMessage another agent's inference returns an assistant message onInternalEvent its own inference emits a semantic stream event onExternalEvent another participant emits a semantic stream event onError one of its own handlers throws onParticipantError another participant's handler throws Every handler defaults to a no-op on BaseParticipant — override only the ones you care about. flowchart LR Human[Participant] -->|"sendMessage(env, text, caller)"| Env(("AgenticEnvironment")) Agent[Participant] -->|"runInference / executeFunctionCall"| Env Observer[Participant] -->|join| Env Env -->|"onMessage / onExternal*"| Human Env -->|"onFunctionCall / onReasoning / …"| Agent Env -->|"onExternal*"| Observer Env -->|"onJoined / onLeft / onParticipant*"| All Loading Non-blocking participants A participant is any subclass of Participant. Use BaseParticipant as a base when you only want to override a few handlers — every handler it defines is a no-op. The role (human, agent, observer) is just which capability functions a participant calls and which handlers it overrides: Role How to build it Human A participant that calls sendMessage(environment, text, caller) Agent A participant that calls runInference(...) and executeFunctionCall(...) Observer A participant that only overrides handlers and never runs inference import { AgenticEnvironment, BaseParticipant, ModelContext, UserMessageItem, runInference, sendMessage, } from "@mozaik-ai/core" const environment = new AgenticEnvironment() const human = new BaseParticipant() class Agent extends BaseParticipant { private readonly context = ModelContext.create("demo") async onMessage(message: string): Promise<void> { this.context.addContextItem(UserMessageItem.create(message)) runInference({ model: "gpt-5.4", context: this.context, caller: this, environment }) } } const agent = new Agent() const observer = new BaseParticipant() human.join(environment) agent.join(environment) observer.join(environment) sendMessage(environment, "Hello", human) Participants react as soon as they join() agentic environment. The environment fans every item out to every subscriber synchronously and without awaiting them, so a slow listener never blocks producers or other listeners. Reactive agent A reactive agent extends BaseParticipant and overrides the handlers it wants to react on. Each handler is already a no-op in the base class, so only the relevant ones need bodies. Capabilities are the free functions runInference and executeFunctionCall — the participant passes itself as caller: import { BaseParticipant, UserMessageItem, FunctionCallItem, FunctionCallOutputItem, ReasoningItem, ModelMessageItem, AgenticEnvironment, ModelContext, ModelName, Tool, runInference, executeFunctionCall, } from "@mozaik-ai/core" export class ReactiveAgent extends BaseParticipant { constructor( private readonly environment: AgenticEnvironment, private readonly context: ModelContext, private readonly tools: Tool[] = [], ) { super() } // A message from a human (or any other participant) → record it and think. async onMessage(message: string): Promise<void> { this.context.addContextItem(UserMessageItem.create(message)) runInference({ model: 'gpt-5.5', context: this.context, tools: this.tools, caller: this, environment: this.environment, }) } // The agent just produced a function call → execute it. async onFunctionCall(item: FunctionCallItem): Promise<void> { this.context.addContextItem(item) const tool = this.tools.find((t) => t.name === item.name) if (tool) executeFunctionCall(this.environment, item, tool, this) } // The tool just produced an output → feed it back and run inference again. async onFunctionCallOutput(item: FunctionCallOutputItem): Promise<void> { this.context.addContextItem(item) runInference({ model: 'gpt-5.5', context: this.context, tools: this.tools, caller: this, environment: this.environment, }) } // Keep the local context in sync with model-emitted reasoning and replies. async onReasoning(item: ReasoningItem): Promise<void> { this.context.addContextItem(item) } async onModelMessage(item: ModelMessageItem): Promise<void> { this.context.addContextItem(item) } } Three things to note: The split between self handlers and onExternal* handlers means a participant can encode "act on my own outputs" separately from "observe others", without inspecting source by hand. The agent never awaits its capability calls inside the handlers — runInference and executeFunctionCall are fire-and-forget (they return void), so the environment keeps delivering events while inference and tool execution run in the background. Behaviors compose by reaction, not orchestration. Add a second agent that overrides onExternalModelMessage and you get a critique loop. Add a TranscriptLogger and you get a UI stream. Neither change touches the existing participants. Streaming and semantic events When inference runs with streaming enabled (streaming: true on the runInference params, for a model that supports it), the runner does not wait for the full response. As the provider emits chunks, the endpoint yields SemanticEvent items (type + data) and the environment delivers each one to every joined participant immediately — the same fan-out as messages and context items. Participants react in real time by overriding the stream handlers; no participant needs to poll or share a callback. The producing participant receives onInternalEvent; everyone else receives onExternalEvent(source, event): import { BaseParticipant, Participant, SemanticEvent } from "@mozaik-ai/core" // Agent that runs streaming inference — can observe its own stream chunks. export class StreamingAgent extends BaseParticipant { async onInternalEvent(event: SemanticEvent<unknown>): Promise<void> { if (event.type === "response.output_text.delta") { // e.g. keep a local buffer of partial output } } } // Any other participant — UI, logger, second agent — reacts to another's stream. export class LiveTranscript extends BaseParticipant { async onExternalEvent(source: Participant, event: SemanticEvent<unknown>): Promise<void> { if (event.type === "response.output_text.delta") { const { delta } = event.data as { delta: string } process.stdout.write(delta) } } } Enable streaming by passing streaming: true to runInference: runInference({ model: "gpt-5.4", context, caller: this, environment, streaming: true }) Requesting streaming for a model whose specification has supportsStreaming: false fails request validation before the API is called. Lifecycle hooks Every participant receives lifecycle notifications when it or others join/leave an environment: export class TeamAgent extends BaseParticipant { // Called when this participant joins an environment. onJoined(): void { console.log("I joined the environment") } // Called when this participant leaves an environment. onLeft(): void { console.log("I left the environment") } // Called when another participant joins the same environment. onParticipantJoined(participant: Participant): void { console.log(`${participant.constructor.name} joined`) } // Called when another participant leaves the same environment. onParticipantLeft(participant: Participant): void { console.log(`${participant.constructor.name} left`) } } This lets participants react to membership changes — for example, an agent could start inference only after a required collaborator has joined, or clean up shared state when someone leaves. Reacting to external events Participants can listen to external events and react by overriding methods like onMessage, onExternalFunctionCall, onExternalFunctionCallOutput, onExternalReasoning, and onExternalModelMessage. Selective listening By default a participant reacts to events from every other participant. To scope a participant so it only reacts to specific participant types, populate its listens list with those classes. When listens is non-empty, the environment only delivers external events whose source is an instance of one of the listed classes: import { BaseParticipant } from "@mozaik-ai/core" export class Critic extends BaseParticipant { // Only react to events produced by Writer participants. protected listens = [Writer] } Error handling When any handler throws, the environment catches it and routes it as an AgenticError instead of crashing the run. The participant whose handler threw receives onError(error); every other participant receives onParticipantError(source, error). After its own onError, the failing participant is marked inactive in that environment so it stops receiving further events. import { BaseParticipant, Participant, AgenticError } from "@mozaik-ai/core" export class ResilientAgent extends BaseParticipant { onError(error: AgenticError): void { console.error("my handler threw:", error.message) } onParticipantError(source: Participant, error: AgenticError): void { console.warn(`${source.constructor.name} failed:`, error.message) } } AgenticError carries the originating participant (getSource()) and environment (getEnvironment()). Passive observer You can create observers that don't run inference themselves but watch what's happening in the conversation and take side actions (logging, metrics, persistence, etc.). Subclass BaseParticipant and override only the handlers you care about — everything else stays a no-op: import { BaseParticipant, Participant, FunctionCallItem, FunctionCallOutputItem, ReasoningItem, ModelMessageItem, } from "@mozaik-ai/core" export class TranscriptLogger extends BaseParticipant { async onMessage(message: string): Promise<void> { console.log("[message]", message) } async onExternalFunctionCall(source: Participant, item: FunctionCallItem): Promise<void> { console.log(`[${source.constructor.name}] function_call`, item) } async onExternalFunctionCallOutput(source: Participant, item: FunctionCallOutputItem): Promise<void> { console.log(`[${source.constructor.name}] function_call_output`, item) } async onExternalReasoning(source: Participant, item: ReasoningItem): Promise<void> { console.log(`[${source.constructor.name}] reasoning`, item) } async onExternalModelMessage(source: Participant, item: ModelMessageItem): Promise<void> { console.log(`[${source.constructor.name}] model_message`, item) } } Examples Working examples are available here: mozaik-examples.
Preview unavailable