A Semantic Kernel Plugin for Live Web Search
Expose Serply's search and scraper endpoints as native Semantic Kernel functions the planner can compose on its own.
Semantic Kernel’s plugin model rewards precision. You decorate methods with @kernel_function, describe them well, and the planner composes them without you writing orchestration code. The catch is that the planner is only as good as your descriptions — it never sees your implementation, only the metadata.
That makes search plugins a good test case. “Searches the web” is nearly useless to a planner. “Returns current web results with titles, URLs, and snippets — use for anything that may have changed since training” gets called at the right moments.
The plugin
import os
import requests
from urllib.parse import quote_plus
from typing import Annotated
from semantic_kernel.functions import kernel_function
BASE = "https://api.serply.io/v1"
class WebSearchPlugin:
"""Live web search and page reading via Serply."""
def __init__(self, api_key: str | None = None, proxy_location: str = "US"):
self.api_key = api_key or os.environ["SERPLY_API_KEY"]
self.proxy_location = proxy_location
@property
def _headers(self) -> dict:
return {"X-Api-Key": self.api_key, "X-Proxy-Location": self.proxy_location}
@kernel_function(
name="search",
description=(
"Search the live web and return ranked results with titles, URLs, "
"and snippets. Use for current events, prices, releases, or any "
"fact that may have changed recently."
),
)
def search(
self,
query: Annotated[str, "The search query, phrased as a Google search"],
num: Annotated[int, "Number of results to return, 1-50"] = 10,
) -> Annotated[str, "Formatted list of search results"]:
resp = requests.get(
f"{BASE}/search/q={quote_plus(query)}&num={num}",
headers=self._headers,
timeout=30,
)
if resp.status_code == 429:
return "Rate limit reached. Do not retry; proceed with available information."
resp.raise_for_status()
results = resp.json().get("results", [])
if not results:
return f"No results for '{query}'."
return "\n\n".join(
f"{r.get('position')}. {r.get('title')}\n{r.get('link')}\n{r.get('description', '')}"
for r in results
)
@kernel_function(
name="read_page",
description=(
"Fetch the full text of a web page as markdown. Use after search "
"when the snippet does not contain enough detail to answer."
),
)
def read_page(
self,
url: Annotated[str, "Full URL of the page to read"],
) -> Annotated[str, "Page content as markdown"]:
resp = requests.post(
f"{BASE}/request",
headers={**self._headers, "Content-Type": "application/json"},
json={"url": url, "response_type": "markdown"},
timeout=60,
)
if resp.status_code == 429:
return "Rate limit reached while reading the page."
resp.raise_for_status()
return resp.text[:15000]Two implementation notes on the Serply side. The search endpoint takes a Google-style query string as a path segment — /v1/search/q=... — rather than as query parameters. And the scraper’s markdown mode returns text in the response body, not JSON, so .text is correct; "response_type": "full" would instead give you JSON with the raw HTML under data.
Returning a plain string on 429 rather than raising is a deliberate choice for planner-driven flows. An exception aborts the plan; a string the planner can read lets it finish with what it has.
Registering and running
import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings
kernel = Kernel()
kernel.add_service(OpenAIChatCompletion(
service_id="chat",
ai_model_id="gpt-4o",
api_key=os.environ["OPENAI_API_KEY"],
))
kernel.add_plugin(WebSearchPlugin(), plugin_name="web")
settings = OpenAIChatPromptExecutionSettings(
service_id="chat",
function_choice_behavior=FunctionChoiceBehavior.Auto(maximum_auto_invoke_attempts=6),
)
async def main():
answer = await kernel.invoke_prompt(
"Which companies announced new SERP API pricing this quarter? Cite URLs.",
arguments_settings=settings,
)
print(answer)
asyncio.run(main())maximum_auto_invoke_attempts is the guardrail that matters. Left unbounded, a model that decides it needs one more search will keep getting one; six is generous for most questions and still bounded.
Adding more surfaces
The same class can carry the other endpoints, and the planner will pick between them if the descriptions are distinct enough:
@kernel_function(
name="search_news",
description=(
"Search recent news articles. Use when the question is about "
"events, announcements, or coverage rather than reference material."
),
)
def search_news(
self, query: Annotated[str, "News search query"]
) -> Annotated[str, "Recent news articles"]:
resp = requests.get(
f"{BASE}/news/q={quote_plus(query)}",
headers=self._headers,
timeout=30,
)
resp.raise_for_status()
entries = resp.json().get("feed", {}).get("entries", [])
return "\n\n".join(
f"{e.get('title')}\n{e.get('source')} — {e.get('published')}\n{e.get('link')}"
for e in entries[:15]
) or "No recent articles found."The news endpoint has a different response shape from search: articles live at feed.entries, each with title, link, summary, published, and source. Getting that wrong is the most common integration bug, because the outer object looks similar enough that a missing results key reads as “no data” rather than “wrong path.”
Practical note on descriptions
If the planner isn’t calling your search function when it obviously should, the fix is almost never a better system prompt. It’s the function description. Name the situations, not the mechanism — planners match on “current events, prices, releases” far more reliably than on “queries a search engine.”