v1.3.0

AG2 1.3 adds observation across delegation trees, explicit variable scopes, dynamic model context, and a shared task lifecycle.

Deprecations and migration to 1.4

The following APIs continue to work in 1.3 with deprecation warnings and will be removed in 1.4.0:

Deprecated API	Replacement and migration guide
TaskConfig, Agent(tasks=...), Agent.tasks, and the automatically injected run_subtask / run_subtasks tools	Construct workers explicitly and delegate through Agent.as_tool(). Migration guide
Agent(assembly=...)	Register AssemblerMiddleware through middleware=; add HaltCheckMiddleware when using AlertPolicy. Assembly guide
Agent(knowledge=...)	Use plugins=[KnowledgePlugin(...)]. Migration guide
Agent.task()	Construct Task(owner_id=..., spec=TaskSpec(...)) directly for standalone work. Task API

KnowledgeConfig, assembly policies, and Task remain supported; the deprecations apply to the entry points above.

As previously announced, OpenAIConfig will become an alias for OpenAIResponsesConfig in 1.4. Use OpenAIResponsesConfig or OpenAICompletionsConfig explicitly. Model configuration.

Python requirement

AG2 1.3 requires Python 3.11 or newer. Upgrade Python before updating an environment that still uses 3.10.

The examples below use OpenAIResponsesConfig. Install ag2[openai] and set OPENAI_API_KEY as described in the Quick Start. Run statements containing await inside an async function. Later fragments reuse config from the first example.

Observe the entire delegation tree

Ancestors now receive events from every descendant, including streamed chunks. Operational events arrive as ChildEvent observations: they do not enter the ancestor's model transcript or trigger its tools, model loop, or HITL handlers.

Typed @observer(...) callbacks let a coordinator monitor the whole tree. An observed event's live event.origin addresses the invocation that authored it; use await event.origin.send(...) to deliver an event back through that invocation's normal handlers. Handles expire when the invocation ends and are not available when replaying saved history.

Documentation: event observation and addressed delivery.

Explicit variable scopes

ScopedValue defines how state crosses a delegation boundary:

Scope	Behavior
COPIED	Deep-copied for each branch; the default for plain values
LOCAL	Not inherited by descendants
SHARED	One binding across the invocation's tree, including replacement of its value

Declare shared names at the root. Applications provide synchronization for concurrent updates; locks, clients, and connections belong in dependencies. Fresh root invocations copy their input values, so read results from reply.context.variables rather than the caller's original containers.

Documentation: variable scopes.

Use event observation and scoped state together:

from ag2 import Agent, Context, Scope, ScopedValue, observer, tool
from ag2.config import OpenAIResponsesConfig
from ag2.events import ChildEvent, ModelMessageChunk

config = OpenAIResponsesConfig(model="gpt-5", streaming=True)

@tool
def save_finding(finding: str, context: Context) -> str:
    """Save a finding and update this worker's draft."""
    context.variables["findings"].append(finding)
    context.variables["draft"].append(finding)
    return "Saved"

@observer(ChildEvent)
def show_progress(event: ChildEvent) -> None:
    source = event.source_event
    if isinstance(source, ModelMessageChunk):
        print(f"{event.agent_name}: {source.content}")

worker = Agent(
    "worker",
    config=config,
    prompt="Call save_finding once to record the requested finding.",
    tools=[save_finding],
)
coordinator = Agent(
    "coordinator",
    config=config,
    prompt="Delegate the task to task_worker once, then summarize its response.",
    tools=[worker.as_tool(description="Record a finding.")],
    observers=[show_progress],
    variables={
        "findings": ScopedValue([], scope=Scope.SHARED),
        "draft": [],
        "root_note": ScopedValue("Coordinator only", scope=Scope.LOCAL),
    },
)

reply = await coordinator.ask(
    "Record that each delegated invocation has its own conversation history."
)
print(reply.context.variables["findings"])  # Finding saved by the worker
print(reply.context.variables["draft"])     # []

Descendant usage arrives while work runs

Original UsageEvent records reach ancestors as they arrive, preserving their source identity, stage, model, and provider. New delegations no longer generate subtask rollups. Existing @observer(UsageEvent) callbacks receive descendant spend, and reports count each source publication once.

Consumers that grouped new spend under kind="subtask" should use the source records and their delegation provenance. Response and terminal task usage snapshots are for inspection, not additional additive records.

Documentation: descendant usage accounting.

report = await reply.usage()
print(report.total.total_tokens)
print(report.by_model)

Dynamic context before each model call

statusbar_middleware presents current facts such as a plan, progress, remaining attempts, or environment state before the model's next decision. Callbacks support dependency injection; persist=True retains snapshots as service observations.

Documentation: Status Bars.

from ag2 import DataInput
from ag2.middleware import statusbar_middleware

@statusbar_middleware(persist=True)
def task_status(context: Context) -> DataInput:
    return DataInput({
        "goal": context.variables["goal"],
        "remaining_steps": context.variables["remaining_steps"],
    })

agent = Agent(
    "writer",
    config=config,
    middleware=[task_status],
    variables={
        "goal": "Prepare release notes",
        "remaining_steps": ["Draft", "Review"],
    },
)

Query retained history

Filter saved events by type and conditions, select the first or last match, or count matching occurrences. since_last_user_input() limits a query to the latest user-input interval. Queries read retained top-level events; records removed by compaction are no longer available.

Documentation: history queries.

from ag2.events import ToolCallEvent

calls = reply.context.stream.history.since_last_user_input().where(ToolCallEvent)
print(await calls.count())
print(await calls.last())

A shared task lifecycle

Model runs, delegations, and standalone Task blocks use the same lifecycle transitions. Enable lifecycle events for ask, run, or resume with capability or ttl_seconds; delegation enables them automatically.

TTL supplies an expiry deadline for an external observer. It does not impose a local execution timeout.

Documentation: Tasks.

reply = await agent.ask(
    "Draft the release notes.",
    capability="writing",
    ttl_seconds=60,
)

Knowledge as a plugin

KnowledgePlugin composes a store, the knowledge tool, bootstrap, and event logging. Configure compaction and aggregation through the plugin, supplying each strategy with its corresponding trigger.

Documentation: KnowledgePlugin.

from ag2.knowledge import KnowledgePlugin, MemoryKnowledgeStore

agent = Agent(
    "assistant",
    config=config,
    plugins=[KnowledgePlugin(store=MemoryKnowledgeStore())],
)

Other improvements

• Mistral realtime TTS sends a reply's text over one connection per reply. Documentation.
• Gemini's automatic_continuation controls resumption of responses that reach the per-request token limit. Documentation.
• MCP's max_sse_event_size configures the maximum accepted event size for tool results. Documentation.
• TrackingConfig captures the complete ordered model input for each call. Documentation.

Fixes

• A2A reconnect recovery, artifact reconciliation, and cancellation racing completion. Client guide.
• Local command cleanup now stops child processes on cancellation, timeout, and close. Local Shell.
• Final tool batches are retained for correct conversation replay and continuation. Agent guide.
• TaskInject resolves the enclosing task correctly under concurrent sibling tasks. Task guide.

Full changelog: v1.2.2...v1.3.0