microsoft/autogen · error · NotImplementedError

Agent save_state is not yet implemented.

Error message

Agent save_state is not yet implemented.

What it means

GrpcWorkerAgentRuntime.agent_save_state() always raises NotImplementedError: saving an individual agent's state via the runtime is not supported in the distributed gRPC model, where agents live in (possibly remote) worker processes the runtime cannot introspect.

Source

Thrown at python/packages/autogen-ext/src/autogen_ext/runtimes/grpc/_worker_runtime.py:502

                )

            telemetry_metadata = get_telemetry_grpc_metadata()
            task = asyncio.create_task(self._send_message(runtime_message, "publish", topic_id, telemetry_metadata))
            self._background_tasks.add(task)
            task.add_done_callback(self._raise_on_exception)
            task.add_done_callback(self._background_tasks.discard)

    async def save_state(self) -> Mapping[str, Any]:
        raise NotImplementedError("Saving state is not yet implemented.")

    async def load_state(self, state: Mapping[str, Any]) -> None:
        raise NotImplementedError("Loading state is not yet implemented.")

    async def agent_metadata(self, agent: AgentId) -> AgentMetadata:
        raise NotImplementedError("Agent metadata is not yet implemented.")

    async def agent_save_state(self, agent: AgentId) -> Mapping[str, Any]:
        raise NotImplementedError("Agent save_state is not yet implemented.")

    async def agent_load_state(self, agent: AgentId, state: Mapping[str, Any]) -> None:
        raise NotImplementedError("Agent load_state is not yet implemented.")

    async def _get_new_request_id(self) -> str:
        async with self._pending_requests_lock:
            self._next_request_id += 1
            return str(self._next_request_id)

    async def _process_request(self, request: agent_worker_pb2.RpcRequest) -> None:
        assert self._host_connection is not None
        recipient = AgentId(request.target.type, request.target.key)
        sender: AgentId | None = None
        if request.HasField("source"):
            sender = AgentId(request.source.type, request.source.key)
            logging.info(f"Processing request from {sender} to {recipient}")
        else:
            logging.info(f"Processing request from unknown source to {recipient}")

View on GitHub (pinned to 027ecf0a37)

Solutions

  1. Persist agent state inside the agent itself (its handlers or a dedicated save message) rather than via the runtime
  2. Condition persistence logic on runtime type and skip for gRPC workers
  3. Use the local runtime where per-agent state APIs are required

Example fix

# before
state = await runtime.agent_save_state(agent_id)  # NotImplementedError

# after
state = await runtime.send_message(SaveStateRequest(), agent_id)  # agent handles it
Defensive patterns

Strategy: type-guard

Validate before calling

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

if isinstance(runtime, GrpcWorkerAgentRuntime):
    state = await runtime.send_message(SaveStateRequest(), agent_id)  # agent handles it
else:
    state = await runtime.agent_save_state(agent_id)

Type guard

def runtime_has_per_agent_state(runtime: object) -> bool:
    return not isinstance(runtime, GrpcWorkerAgentRuntime)

Try / catch

try:
    state = await runtime.agent_save_state(agent_id)
except NotImplementedError:
    state = await runtime.send_message(SaveStateRequest(), agent_id)

Prevention

When it happens

Trigger: Calling `await runtime.agent_save_state(agent_id)` on GrpcWorkerAgentRuntime, e.g. from generic persistence layers built on the AgentRuntime interface.

Common situations: Migrating stateful multi-agent apps from the local runtime; framework code that snapshots each agent via the runtime API.

Related errors

AI-assisted analysis of microsoft/autogen@027ecf0a37 (2026-08-15). Data as JSON: /api/errors/16fc4bb4b7873c4d. Report an issue: GitHub.