microsoft/autogen · error · NotImplementedError

Agent metadata is not yet implemented.

Error message

Agent metadata is not yet implemented.

What it means

GrpcWorkerAgentRuntime.agent_metadata() is a stub raising NotImplementedError: per-agent metadata queries are not implemented in the distributed worker runtime (metadata is a concept of the local agent runtime's direct-instantiation model).

Source

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

                        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)
        sender: AgentId | None = None
        if request.HasField("source"):
            sender = AgentId(request.source.type, request.source.key)

View on GitHub (pinned to 027ecf0a37)

Solutions

  1. Remove or condition the agent_metadata call for gRPC runtimes
  2. Track agent metadata in your own registry keyed by AgentId for distributed deployments
  3. Gate on runtime type before querying metadata

Example fix

# before
meta = await runtime.agent_metadata(agent_id)  # NotImplementedError

# after
meta = my_metadata_registry.get(agent_id)  # app-managed
Defensive patterns

Strategy: type-guard

Validate before calling

from autogen_ext.runtimes.grpc import GrpcWorkerAgentRuntime

if isinstance(runtime, GrpcWorkerAgentRuntime):
    meta = my_metadata_registry.get(agent_id)
else:
    meta = await runtime.agent_metadata(agent_id)

Type guard

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

Try / catch

try:
    meta = await runtime.agent_metadata(agent_id)
except NotImplementedError:
    meta = my_metadata_registry.get(agent_id)

Prevention

When it happens

Trigger: Calling `await runtime.agent_metadata(agent_id)` on GrpcWorkerAgentRuntime, often indirectly through helper code that queries metadata for logging or UI display.

Common situations: Dashboards/telemetry code written against the AgentRuntime interface being pointed at a gRPC worker; shared utilities that call agent_metadata on any runtime.

Related errors

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