microsoft/autogen · error · NotImplementedError

Loading state is not yet implemented.

Error message

Loading state is not yet implemented.

What it means

GrpcWorkerAgentRuntime.load_state() is a stub that always raises NotImplementedError, mirroring save_state: the distributed runtime does not support restoring global runtime state from a mapping.

Source

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

                        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)

    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)

View on GitHub (pinned to 027ecf0a37)

Solutions

  1. Drop runtime-level load_state for gRPC deployments and restore state inside your agents' own handlers or constructor-time hydration
  2. Keep a local SingleThreadedAgentRuntime variant for stateful tests/simulations
  3. Feature-detect before calling: hasattr/runtime type checks to skip persistence for gRPC runtimes

Example fix

# before
await runtime.load_state(saved)  # NotImplementedError

# after
if isinstance(runtime, GrpcWorkerAgentRuntime):
    await hydrate_agents_from_store()
else:
    await runtime.load_state(saved)
Defensive patterns

Strategy: type-guard

Validate before calling

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

if isinstance(runtime, GrpcWorkerAgentRuntime):
    await hydrate_agents_from_store()
else:
    await runtime.load_state(saved)

Type guard

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

Try / catch

try:
    await runtime.load_state(state)
except NotImplementedError:
    await hydrate_agents_from_store()  # app-level restore

Prevention

When it happens

Trigger: Calling `await runtime.load_state(state)` on GrpcWorkerAgentRuntime, typically from generic restore/checkpoint code paths that accept any AgentRuntime.

Common situations: Porting a SingleThreadedAgentRuntime app with save/restore to gRPC workers; framework checkpoint tools invoking load_state uniformly.

Related errors

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