blackboard_query
Query beliefs stored in a shared blackboard using regex patterns on keys or string values, optionally filtered by tag.
Instructions
Query beliefs by regex pattern (over key or string value) and/or tag.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| pattern | No | ||
| tag | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The FastMCP tool handler for blackboard_query. Registered via @mcp.tool() decorator. Delegates to Blackboard.query() and returns results as dicts.
@mcp.tool() def blackboard_query( pattern: str | None = None, tag: str | None = None ) -> list[dict[str, Any]]: """Query beliefs by regex pattern (over key or string value) and/or tag.""" return [b.to_dict() for b in _BLACKBOARD.query(pattern, tag=tag)] - Input schema: pattern (optional str regex) and tag (optional str). Output: list of dicts with belief fields.
@mcp.tool() def blackboard_query( pattern: str | None = None, tag: str | None = None ) -> list[dict[str, Any]]: """Query beliefs by regex pattern (over key or string value) and/or tag.""" return [b.to_dict() for b in _BLACKBOARD.query(pattern, tag=tag)] - src/mcp_research_collective/mcp_server.py:63-63 (registration)Tool registered with the FastMCP server via the @mcp.tool() decorator on line 63.
@mcp.tool() - Blackboard.query() method: the core logic that filters beliefs by regex pattern (matching key or string value) and/or tag, skipping expired entries, returning sorted by timestamp.
def query( self, pattern: Optional[str] = None, *, tag: Optional[str] = None ) -> list[Belief]: with self._lock: out: list[Belief] = [] for b in self._store.values(): if b.expired(): continue if pattern: matches_key = re.search(pattern, b.key, re.I) matches_value = ( isinstance(b.value, str) and re.search(pattern, b.value, re.I) ) if not (matches_key or matches_value): continue if tag and tag not in b.tags: continue out.append(b) return sorted(out, key=lambda x: x.ts) - MCPConnector.call() handles 'blackboard.query' tool name routing, used by agent code (e.g., in agents.py).
if tool == "blackboard.query": results = self.blackboard.query( kwargs.get("pattern"), tag=kwargs.get("tag") ) return [b.to_dict() for b in results] raise NotImplementedError(f"Unknown tool: {tool}")