microsoft/autogen · error · NotImplementedError

Agent load_state is not yet implemented.

Error message

Agent load_state is not yet implemented.

What it means

GrpcWorkerAgentRuntime.agent_load_state() is a stub that raises NotImplementedError: restoring an individual agent's state through the runtime is not supported for distributed workers, symmetric with agent_save_state.

Source

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

            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}")

        # Deserialize the message.
        message = self._serialization_registry.deserialize(

View on GitHub (pinned to 027ecf0a37)

Solutions

  1. Send a custom LoadState message to the agent and let it restore itself
  2. Hydrate agent state at construction/registration time from your own store
  3. Skip runtime-level restore for gRPC runtimes via a type check

Example fix

# before
await runtime.agent_load_state(agent_id, state)  # NotImplementedError

# after
await runtime.send_message(LoadStateRequest(state), agent_id)  # agent applies it
Defensive patterns

Strategy: type-guard

Validate before calling

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

if isinstance(runtime, GrpcWorkerAgentRuntime):
    await runtime.send_message(LoadStateRequest(state), agent_id)
else:
    await runtime.agent_load_state(agent_id, state)

Type guard

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

Try / catch

try:
    await runtime.agent_load_state(agent_id, state)
except NotImplementedError:
    await runtime.send_message(LoadStateRequest(state), agent_id)

Prevention

When it happens

Trigger: Calling `await runtime.agent_load_state(agent_id, state)` on GrpcWorkerAgentRuntime, usually from generic restore/checkpoint code.

Common situations: Restore logic written for SingleThreadedAgentRuntime reused against gRPC workers; resuming conversations by reloading agent state through the runtime API.

Related errors

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