microsoft/semantic-kernel · error · ValidationError

receive_task must be an instance of asyncio.Task or None

Error message

receive_task must be an instance of asyncio.Task or None

What it means

QueueReference is a pydantic model that owns a receive_task field used to track the background receive coroutine for a broadcast channel. A before-mode model_validator asserts that receive_task, if not None, is an instance of asyncio.Task. This prevents storing bare coroutines or Futures that would never be awaited/cancelled correctly by the broadcast loop.

Source

Thrown at python/semantic_kernel/agents/group_chat/broadcast_queue.py:37

    """Utility class to associate a queue with its specific lock."""

    queue: deque = Field(default_factory=deque)
    queue_lock: SkipValidation[asyncio.Lock] = Field(default_factory=asyncio.Lock, exclude=True)
    receive_task: SkipValidation[asyncio.Task | None] = None
    receive_failure: Exception | None = None

    @property
    def is_empty(self):
        """Check if the queue is empty."""
        return len(self.queue) == 0

    @model_validator(mode="before")
    def validate_receive_task(cls, values: Any):
        """Validate the receive task."""
        if isinstance(values, dict):
            receive_task = values.get("receive_task")
            if receive_task is not None and not isinstance(receive_task, asyncio.Task):
                raise ValidationError("receive_task must be an instance of asyncio.Task or None")
        return values


@experimental
@dataclass
class ChannelReference:
    """Tracks a channel along with its hashed key."""

    hash: str
    channel: AgentChannel = field(default_factory=AgentChannel)


@experimental
class BroadcastQueue(KernelBaseModel):
    """A queue for broadcasting messages to listeners."""

    queues: dict[str, QueueReference] = Field(default_factory=dict)
    block_duration: float = 0.1

View on GitHub (pinned to c028a0c7dc)

Solutions

1. Leave receive_task unset/None when constructing QueueReference; the broadcast loop assigns the Task itself via asyncio.create_task.
2. If you must supply it, wrap the coroutine first: receive_task=asyncio.create_task(self.receive(channel_ref, queue_ref)).
3. Never pass a coroutine object or a Future; only asyncio.Task (or None) is accepted.

Example fix

// before
QueueReference(queue=..., receive_task=self.receive(ref, q))  # coroutine, not a Task

// after
QueueReference(queue=..., receive_task=None)
# the broadcast loop creates the task itself via asyncio.create_task

import asyncio

def coerce_receive_task(value):
    if value is None or isinstance(value, asyncio.Task):
        return value
    if asyncio.iscoroutine(value):
        return asyncio.create_task(value)
    raise TypeError("receive_task must be asyncio.Task or None")

import asyncio

def is_valid_receive_task(v) -> bool:
    return v is None or isinstance(v, asyncio.Task)

When it happens

Trigger: Constructing QueueReference (directly or via an ORM/serialization layer) with a dict where receive_task holds something other than None or an asyncio.Task — e.g. a raw coroutine object (async def result not wrapped), an asyncio.Future, a concurrent.futures.Future, or a thread handle.

Common situations: Manually instantiating QueueReference in tests or a custom broadcast implementation and assigning `receive_task=some_coro()` instead of `asyncio.create_task(some_coro())`; migrating code that previously used Futures; deserializing queue state from a format that does not preserve the Task type.

Related errors

• Only FunctionChoiceType.AUTO is supported.
• Failed to create Copilot Studio Agent settings: {exc}
• Failed to create Azure OpenAI settings: {exc}
• {self!r} is bound to a different event loop
• task_done() called too many times

AI-assisted analysis of microsoft/semantic-kernel@c028a0c7dc (2026-08-13). Data as JSON: /api/errors/d5075aea5cdb0125. Report an issue: GitHub.