PerplexitySearchToolkit

def search(
    self,
    *,
    max_results: int | Variable | None = None,
    max_tokens_per_page: int | Variable | None = None,
    search_domain_filter: Sequence[str] | Variable | None = None,
    search_recency_filter: RecencyFilter | Variable | None = None,
    search_after_date_filter: str | Variable | None = None,
    search_before_date_filter: str | Variable | None = None,
    name: str = "perplexity_search",
    description: str = (
        "Search the web with the Perplexity Search API. Returns ranked results "
        "with title, URL, snippet, and date. Supports domain allow/deny filters "
        "(prefix a domain with '-' to exclude it) and recency / date-range filters."
    ),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    client = self._client

    @tool(name=name, description=description, middleware=middleware)
    def perplexity_search(
        query: Annotated[str, Field(description="The search query string.")],
        ctx: Context,
    ) -> ToolResult:
        """Run a Perplexity Search API call and return structured results."""
        params: dict[str, Any] = {
            "max_results": resolve_variable(max_results, ctx, param_name="max_results"),
            "max_tokens_per_page": resolve_variable(max_tokens_per_page, ctx, param_name="max_tokens_per_page"),
            "search_domain_filter": resolve_variable(search_domain_filter, ctx, param_name="search_domain_filter"),
            "search_recency_filter": resolve_variable(
                search_recency_filter, ctx, param_name="search_recency_filter"
            ),
            "search_after_date_filter": resolve_variable(
                search_after_date_filter, ctx, param_name="search_after_date_filter"
            ),
            "search_before_date_filter": resolve_variable(
                search_before_date_filter, ctx, param_name="search_before_date_filter"
            ),
        }
        kwargs = {k: v for k, v in params.items() if v is not None}

        raw = client.search.create(query=query, **kwargs)

        raw_results: list[SearchAPIResult] = getattr(raw, "results", None) or []
        results = [
            PerplexitySearchResult(
                title=r.title or "",
                url=r.url or "",
                snippet=r.snippet,
                date=r.date,
            )
            for r in raw_results
        ]

        return ToolResult(PerplexitySearchResponse(query=query, results=results))

    return perplexity_search

def answer(
    self,
    *,
    model: SonarModel | Variable | None = None,
    max_tokens: int | Variable | None = None,
    search_domain_filter: Sequence[str] | Variable | None = None,
    search_context_size: SearchContextSize | Variable | None = None,
    search_mode: SearchMode | Variable | None = None,
    search_recency_filter: RecencyFilter | Variable | None = None,
    return_images: bool | Variable | None = None,
    return_related_questions: bool | Variable | None = None,
    name: str = "perplexity_answer",
    description: str = (
        "Ask Perplexity AI for an LLM-generated answer with web citations. "
        "Useful for conversational questions, in-depth research, and analysis. "
        "Returns content text plus ranked search results and citations."
    ),
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    client = self._client

    @tool(name=name, description=description, middleware=middleware)
    def perplexity_answer(
        query: Annotated[str, Field(description="The question to answer.")],
        ctx: Context,
    ) -> ToolResult:
        """Generate a Perplexity Sonar answer with citations and search results."""
        resolved_model = resolve_variable(model, ctx, param_name="model") or "sonar"
        resolved_max_tokens = resolve_variable(max_tokens, ctx, param_name="max_tokens") or 1000
        resolved_context_size = (
            resolve_variable(search_context_size, ctx, param_name="search_context_size") or "high"
        )

        params: dict[str, Any] = {
            "search_domain_filter": resolve_variable(search_domain_filter, ctx, param_name="search_domain_filter"),
            "search_mode": resolve_variable(search_mode, ctx, param_name="search_mode"),
            "search_recency_filter": resolve_variable(
                search_recency_filter, ctx, param_name="search_recency_filter"
            ),
            "return_images": resolve_variable(return_images, ctx, param_name="return_images"),
            "return_related_questions": resolve_variable(
                return_related_questions, ctx, param_name="return_related_questions"
            ),
        }
        kwargs = {k: v for k, v in params.items() if v is not None}

        raw = client.chat.completions.create(
            model=resolved_model,
            messages=[
                {"role": "system", "content": "Be precise and concise."},
                {"role": "user", "content": query},
            ],
            max_tokens=resolved_max_tokens,
            web_search_options={"search_context_size": resolved_context_size},
            **kwargs,
        )

        content: str | None = None
        choices = getattr(raw, "choices", None) or []
        if choices:
            message = getattr(choices[0], "message", None)
            content = getattr(message, "content", None) if message is not None else None

        search_results: list[APIPublicSearchResult] = getattr(raw, "search_results", None) or []
        results = [
            PerplexitySearchResult(
                title=r.title or "",
                url=r.url or "",
                snippet=r.snippet,
                date=r.date,
            )
            for r in search_results
        ]

        citations = list(getattr(raw, "citations", None) or [])

        raw_images: list[dict[str, Any]] = list(getattr(raw, "images", None) or [])
        images = [
            PerplexityImageMeta(
                image_url=url,
                origin_url=img.get("origin_url"),
                title=img.get("title"),
                width=img.get("width"),
                height=img.get("height"),
            )
            for img in raw_images
            if (url := img.get("image_url"))
        ]

        response = PerplexitySearchResponse(
            query=query,
            results=results,
            content=content,
            citations=citations,
            images=images,
        )

        image_parts = [ImageInput(url=img.image_url) for img in images]
        return ToolResult(response, *image_parts)

    return perplexity_answer

def set_provider(self, provider: Provider) -> None:
    for t in self.tools:
        t.set_provider(provider)

async def schemas(self, context: "Context") -> Iterable[ToolSchema]:
    schemas: list[ToolSchema] = []
    for t in self.tools:
        schemas.extend(await t.schemas(context))
    return schemas

def register(
    self,
    stack: "ExitStack",
    context: "Context",
    *,
    middleware: Iterable["BaseMiddleware"] = (),
) -> None:
    for t in self.tools:
        t.register(stack, context, middleware=middleware)

def tool(
    self,
    function: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    schema: FunctionParameters | None = None,
    sync_to_thread: bool = True,
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    def make_tool(f: Callable[..., Any]) -> FunctionTool:
        t = tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self._add_tool(t)
        return t

    if function:
        return make_tool(function)

    return make_tool