Skip to content

agent_k.toolsets.search

Web and academic search toolset.

agent_k.toolsets.search

Search tool helpers for AGENT-K agents.

@notice: | Search tool helpers for AGENT-K agents.

@dev: | See module for implementation details and extension points.

@graph: id: agent_k.toolsets.search provides: - agent_k.toolsets.search:build_kaggle_search_query - agent_k.toolsets.search:build_scholarly_query - agent_k.toolsets.search:create_web_search_tool - agent_k.toolsets.search:prepare_web_search - agent_k.toolsets.search:create_web_fetch_tool - agent_k.toolsets.search:prepare_web_fetch pattern: toolset

@similar: - id: agent_k.toolsets.browser when: "Use for browsing fetched pages rather than constructing queries."

@agent-guidance: do: - "Use agent_k.toolsets.search as the canonical home for this capability." do_not: - "Create parallel modules without updating @similar or @graph."

@human-review: last-verified: 2026-01-26 owners: - agent-k-core

(c) Mike Casale 2025. Licensed under the MIT License.

build_kaggle_search_query

build_kaggle_search_query(
    query: Annotated[str, Doc("Search query text.")],
) -> str

Build a Kaggle-scoped web search query.

@notice: | Prefixes the query with site restrictions for Kaggle.

Source code in agent_k/toolsets/search.py
63
64
65
66
67
68
69
def build_kaggle_search_query(query: Annotated[str, Doc("Search query text.")]) -> str:
    """Build a Kaggle-scoped web search query.

    @notice: |
        Prefixes the query with site restrictions for Kaggle.
    """
    return f"site:kaggle.com {query}".strip()

build_scholarly_query

build_scholarly_query(
    topic: Annotated[str, Doc("Topic to search for.")],
    source: Annotated[
        str,
        Doc(
            "Source filter: arxiv, paperswithcode, or all."
        ),
    ] = "all",
) -> str

Build a web search query for academic sources.

@notice: | Adds site restrictions for scholarly sources.

Source code in agent_k/toolsets/search.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def build_scholarly_query(
    topic: Annotated[str, Doc("Topic to search for.")],
    source: Annotated[str, Doc("Source filter: arxiv, paperswithcode, or all.")] = "all",
) -> str:
    """Build a web search query for academic sources.

    @notice: |
        Adds site restrictions for scholarly sources.
    """
    if source == "arxiv":
        return f"site:arxiv.org {topic}".strip()
    if source == "paperswithcode":
        return f"site:paperswithcode.com {topic}".strip()
    return f"site:arxiv.org OR site:paperswithcode.com {topic}".strip()

create_web_search_tool

create_web_search_tool(
    *,
    search_context_size: Annotated[
        Literal["low", "medium", "high"],
        Doc("Search context size."),
    ] = "medium",
    user_location: Annotated[
        WebSearchUserLocation | None,
        Doc("Optional user location context."),
    ] = None,
    blocked_domains: Annotated[
        list[str] | None,
        Doc("Domains to exclude from search."),
    ] = None,
    allowed_domains: Annotated[
        list[str] | None,
        Doc("Domains to allow for search."),
    ] = None,
    max_uses: Annotated[
        int | None, Doc("Maximum tool uses per run.")
    ] = None,
) -> WebSearchTool

Create a WebSearchTool with explicit configuration.

@notice: | Builds a configured WebSearchTool instance.

Source code in agent_k/toolsets/search.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def create_web_search_tool(
    *,
    search_context_size: Annotated[Literal["low", "medium", "high"], Doc("Search context size.")] = "medium",
    user_location: Annotated[WebSearchUserLocation | None, Doc("Optional user location context.")] = None,
    blocked_domains: Annotated[list[str] | None, Doc("Domains to exclude from search.")] = None,
    allowed_domains: Annotated[list[str] | None, Doc("Domains to allow for search.")] = None,
    max_uses: Annotated[int | None, Doc("Maximum tool uses per run.")] = None,
) -> WebSearchTool:
    """Create a WebSearchTool with explicit configuration.

    @notice: |
        Builds a configured WebSearchTool instance.
    """
    return WebSearchTool(
        search_context_size=search_context_size,
        user_location=user_location,
        blocked_domains=blocked_domains,
        allowed_domains=allowed_domains,
        max_uses=max_uses,
    )
prepare_web_search(
    ctx: Annotated[
        RunContext[Any],
        Doc("Run context for tool preparation."),
    ],
) -> WebSearchTool | None

Prepare WebSearchTool dynamically based on RunContext.

@notice: | Returns a WebSearchTool when provider and deps allow.

Source code in agent_k/toolsets/search.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def prepare_web_search(
    ctx: Annotated[RunContext[Any], Doc("Run context for tool preparation.")],
) -> WebSearchTool | None:
    """Prepare WebSearchTool dynamically based on RunContext.

    @notice: |
        Returns a WebSearchTool when provider and deps allow.
    """
    if ctx.model.system not in {"anthropic", "openai", "google", "groq"}:
        return None
    if OpenAIChatModel is not None and isinstance(ctx.model, OpenAIChatModel):
        return None
    if getattr(ctx.deps, "offline_mode", False):
        return None

    user_location = _coerce_user_location(getattr(ctx.deps, "user_location", None))
    blocked_domains = getattr(ctx.deps, "blocked_domains", None)
    allowed_domains = getattr(ctx.deps, "allowed_domains", None)
    max_uses = getattr(ctx.deps, "search_budget", None)

    return create_web_search_tool(
        user_location=user_location, blocked_domains=blocked_domains, allowed_domains=allowed_domains, max_uses=max_uses
    )

create_web_fetch_tool

create_web_fetch_tool(
    *,
    allowed_domains: Annotated[
        list[str] | None, Doc("Domains allowed for fetch.")
    ] = None,
    blocked_domains: Annotated[
        list[str] | None, Doc("Domains blocked for fetch.")
    ] = None,
    max_uses: Annotated[
        int | None, Doc("Maximum tool uses per run.")
    ] = None,
    enable_citations: Annotated[
        bool, Doc("Whether to include citations.")
    ] = True,
    max_content_tokens: Annotated[
        int | None,
        Doc("Maximum tokens for fetched content."),
    ] = None,
) -> WebFetchTool

Create a WebFetchTool with explicit configuration.

@notice: | Builds a configured WebFetchTool instance.

Source code in agent_k/toolsets/search.py
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def create_web_fetch_tool(
    *,
    allowed_domains: Annotated[list[str] | None, Doc("Domains allowed for fetch.")] = None,
    blocked_domains: Annotated[list[str] | None, Doc("Domains blocked for fetch.")] = None,
    max_uses: Annotated[int | None, Doc("Maximum tool uses per run.")] = None,
    enable_citations: Annotated[bool, Doc("Whether to include citations.")] = True,
    max_content_tokens: Annotated[int | None, Doc("Maximum tokens for fetched content.")] = None,
) -> WebFetchTool:
    """Create a WebFetchTool with explicit configuration.

    @notice: |
        Builds a configured WebFetchTool instance.
    """
    return WebFetchTool(
        allowed_domains=allowed_domains,
        blocked_domains=blocked_domains,
        max_uses=max_uses,
        enable_citations=enable_citations,
        max_content_tokens=max_content_tokens,
    )

prepare_web_fetch async

prepare_web_fetch(
    ctx: Annotated[
        RunContext[Any],
        Doc("Run context for tool preparation."),
    ],
) -> WebFetchTool | None

Prepare WebFetchTool dynamically based on RunContext.

@notice: | Returns a WebFetchTool when provider and deps allow.

Source code in agent_k/toolsets/search.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def prepare_web_fetch(
    ctx: Annotated[RunContext[Any], Doc("Run context for tool preparation.")],
) -> WebFetchTool | None:
    """Prepare WebFetchTool dynamically based on RunContext.

    @notice: |
        Returns a WebFetchTool when provider and deps allow.
    """
    if ctx.model.system not in {"anthropic", "google"}:
        return None
    if getattr(ctx.deps, "offline_mode", False):
        return None

    allowed_domains = getattr(ctx.deps, "allowed_domains", None)
    blocked_domains = getattr(ctx.deps, "blocked_domains", None)
    max_uses = getattr(ctx.deps, "fetch_budget", None)

    return create_web_fetch_tool(allowed_domains=allowed_domains, blocked_domains=blocked_domains, max_uses=max_uses)