feat(backend/copilot): add direct ID lookup to find_agent and find_block tools

Enable copilot tools to look up agents by creator/slug and blocks by UUID
directly, avoiding unnecessary search queries when the user provides an
exact identifier. This improves accuracy and reduces latency for direct
lookups.
This commit is contained in:
Zamil Majdy
2026-03-17 12:48:55 +07:00
parent 048fb06b0a
commit f2676de9d0
3 changed files with 117 additions and 21 deletions

View File

@@ -29,6 +29,9 @@ _UUID_PATTERN = re.compile(
re.IGNORECASE,
)
# Matches "creator/slug" identifiers used in the marketplace
_CREATOR_SLUG_PATTERN = re.compile(r"^[\w-]+/[\w-]+$")
# Keywords that should be treated as "list all" rather than a literal search
_LIST_ALL_KEYWORDS = frozenset({"all", "*", "everything", "any", ""})
@@ -72,23 +75,41 @@ async def search_agents(
agents: list[AgentInfo] = []
try:
if source == "marketplace":
logger.info(f"Searching marketplace for: {query}")
results = await store_db().get_store_agents(search_query=query, page_size=5)
for agent in results.agents:
agents.append(
AgentInfo(
id=f"{agent.creator}/{agent.slug}",
name=agent.agent_name,
description=agent.description or "",
source="marketplace",
in_library=False,
creator=agent.creator,
category="general",
rating=agent.rating,
runs=agent.runs,
is_featured=False,
)
# Direct lookup if query matches "creator/slug" pattern
if _CREATOR_SLUG_PATTERN.match(query):
creator, slug = query.split("/", 1)
logger.info(
"Query looks like creator/slug, trying direct lookup: %s",
query,
)
agent_info = await _get_marketplace_agent_by_slug(creator, slug)
if agent_info:
agents.append(agent_info)
logger.info(
"Found marketplace agent by direct lookup: %s",
agent_info.name,
)
if not agents:
logger.info("Searching marketplace for: %s", query)
results = await store_db().get_store_agents(
search_query=query, page_size=5
)
for agent in results.agents:
agents.append(
AgentInfo(
id=f"{agent.creator}/{agent.slug}",
name=agent.agent_name,
description=agent.description or "",
source="marketplace",
in_library=False,
creator=agent.creator,
category="general",
rating=agent.rating,
runs=agent.runs,
is_featured=False,
)
)
else:
if _is_uuid(query):
logger.info(f"Query looks like UUID, trying direct lookup: {query}")
@@ -214,6 +235,37 @@ def _library_agent_to_info(agent: LibraryAgent) -> AgentInfo:
)
async def _get_marketplace_agent_by_slug(creator: str, slug: str) -> AgentInfo | None:
"""Fetch a marketplace agent by creator/slug identifier."""
try:
details = await store_db().get_store_agent_details(creator, slug)
return AgentInfo(
id=f"{details.creator}/{details.slug}",
name=details.agent_name,
description=details.description or "",
source="marketplace",
in_library=False,
creator=details.creator,
category="general",
rating=details.rating,
runs=details.runs,
is_featured=False,
)
except NotFoundError:
logger.debug("Marketplace agent not found: %s/%s", creator, slug)
except DatabaseError:
raise
except Exception as e:
logger.warning(
"Could not fetch marketplace agent %s/%s: %s",
creator,
slug,
e,
exc_info=True,
)
return None
async def _get_library_agent_by_id(user_id: str, agent_id: str) -> AgentInfo | None:
"""Fetch a library agent by ID (library agent ID or graph_id).

View File

@@ -19,7 +19,8 @@ class FindAgentTool(BaseTool):
@property
def description(self) -> str:
return (
"Discover agents from the marketplace based on capabilities and user needs."
"Discover agents from the marketplace based on capabilities and "
"user needs, or look up a specific agent by its creator/slug ID."
)
@property
@@ -29,7 +30,7 @@ class FindAgentTool(BaseTool):
"properties": {
"query": {
"type": "string",
"description": "Search query describing what the user wants to accomplish. Use single keywords for best results.",
"description": "Search query describing what the user wants to accomplish, or a creator/slug ID (e.g. 'username/agent-name') for direct lookup. Use single keywords for best results.",
},
},
"required": ["query"],

View File

@@ -1,4 +1,5 @@
import logging
import re
from typing import Any
from prisma.enums import ContentType
@@ -18,6 +19,11 @@ from .models import (
logger = logging.getLogger(__name__)
_UUID_PATTERN = re.compile(
r"^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$",
re.IGNORECASE,
)
_TARGET_RESULTS = 10
# Over-fetch to compensate for post-hoc filtering of graph-only blocks.
# 40 is 2x current removed; speed of query 10 vs 40 is minimial
@@ -52,7 +58,8 @@ class FindBlockTool(BaseTool):
@property
def description(self) -> str:
return (
"Search for available blocks by name or description. "
"Search for available blocks by name or description, or look up a "
"specific block by its ID. "
"Blocks are reusable components that perform specific tasks like "
"sending emails, making API calls, processing text, etc. "
"IMPORTANT: Use this tool FIRST to get the block's 'id' before calling run_block. "
@@ -68,7 +75,8 @@ class FindBlockTool(BaseTool):
"query": {
"type": "string",
"description": (
"Search query to find blocks by name or description. "
"Search query to find blocks by name or description, "
"or a block ID (UUID) for direct lookup. "
"Use keywords like 'email', 'http', 'text', 'ai', etc."
),
},
@@ -113,11 +121,46 @@ class FindBlockTool(BaseTool):
if not query:
return ErrorResponse(
message="Please provide a search query",
message="Please provide a search query or block ID",
session_id=session_id,
)
try:
# Direct ID lookup if query looks like a UUID
if _UUID_PATTERN.match(query):
block = get_block(query)
if block and not block.disabled:
if (
block.block_type not in COPILOT_EXCLUDED_BLOCK_TYPES
and block.id not in COPILOT_EXCLUDED_BLOCK_IDS
):
summary = BlockInfoSummary(
id=query,
name=block.name,
description=(
block.optimized_description or block.description or ""
),
categories=[c.value for c in block.categories],
)
if include_schemas:
info = block.get_info()
summary.input_schema = info.inputSchema
summary.output_schema = info.outputSchema
summary.static_output = info.staticOutput
return BlockListResponse(
message=(
f"Found block '{block.name}' by ID. "
"To see inputs/outputs and execute it, use "
"run_block with the block's 'id' - providing "
"no inputs."
),
blocks=[summary],
count=1,
query=query,
session_id=session_id,
)
# Search for blocks using hybrid search
results, total = await search().unified_hybrid_search(
query=query,