microsoft/autogen · error · NotImplementedError

Saving state is not yet implemented.

Error message

Saving state is not yet implemented.

What it means

GrpcWorkerAgentRuntime does not implement global state persistence: save_state() unconditionally raises NotImplementedError. In the distributed gRPC design, runtime state lives with agents, so snapshotting the whole runtime is intentionally unimplemented.

Source

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

                runtime_message = agent_worker_pb2.Message(
                    cloudEvent=cloudevent_pb2.CloudEvent(
                        id=message_id,
                        spec_version="1.0",
                        type=topic_id.type,
                        source=topic_id.source,
                        attributes=attributes,
                        proto_data=any_proto,
                    )
                )

            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)

View on GitHub (pinned to 027ecf0a37)

Solutions

  1. Remove runtime-level save_state usage with GrpcWorkerAgentRuntime; persist agent state through your own storage in each agent's handlers
  2. If you need checkpointing, serialize the relevant application state yourself before stopping workers
  3. Use SingleThreadedAgentRuntime for local scenarios that require save_state/load_state

Example fix

# before
state = await runtime.save_state()  # NotImplementedError

# after (persist within your agents / app layer)
await my_store.put("checkpoint", my_app_state)
Defensive patterns

Strategy: type-guard

Validate before calling

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

if isinstance(runtime, GrpcWorkerAgentRuntime):
    raise NotImplementedError("use app-level persistence for gRPC runtimes")
state = await runtime.save_state()

Type guard

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

Try / catch

try:
    state = await runtime.save_state()
except NotImplementedError:
    state = None  # gRPC runtime: persist via app-level storage instead

Prevention

When it happens

Trigger: Calling `await runtime.save_state()` on a GrpcWorkerAgentRuntime, directly or via generic framework code (e.g. checkpointing helpers) that calls save_state on any AgentRuntime.

Common situations: Reusing checkpoint/restore code written for SingleThreadedAgentRuntime with the gRPC runtime; migrating an app to distributed workers while keeping whole-runtime state snapshots.

Related errors

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