Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 112 additions & 3 deletions src/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -281,14 +281,18 @@ async def search(params: dict[str, Any] = None, mode: str = "complete") -> str:
mode: Response mode (default: "complete")
- "complete": Returns full JSON response with all fields
- "compact": Returns JSON response with metadata fields removed
- "formatted": Returns a human-readable text rendering of the
highest-signal parts (answer box, knowledge graph, organic
results, related questions, related searches). Useful for
LLM agents that prefer prose over raw JSON. Experimental.

Returns:
A JSON string containing search results or an error message.
"""

# Validate mode parameter
if mode not in ["complete", "compact"]:
return "Error: Invalid mode. Must be 'complete' or 'compact'"
if mode not in ["complete", "compact", "formatted"]:
return "Error: Invalid mode. Must be 'complete', 'compact', or 'formatted'"

if params is None:
params = {}
Expand Down Expand Up @@ -321,7 +325,10 @@ async def search(params: dict[str, Any] = None, mode: str = "complete") -> str:
for field in fields_to_remove:
data.pop(field, None)

# Return JSON response for both modes
# Return JSON response for "complete" and "compact" modes.
# For "formatted", render the highest-signal parts as text.
if mode == "formatted":
return _format_response_text(data)
return json.dumps(data, indent=2, ensure_ascii=False)

except serpapi.exceptions.HTTPError as e:
Expand Down Expand Up @@ -373,5 +380,107 @@ def main():
uvicorn.run(starlette_app, host=host, port=port, ws="none")


def _truncate(s: str, n: int) -> str:
"""Truncate a string to n chars, appending a marker if cut."""
if len(s) <= n:
return s
return s[:n] + f"… [+{len(s) - n} chars truncated]"


def _format_response_text(data: dict) -> str:
"""Render the highest-signal parts of a SerpApi response as text.

Used by mode="formatted". Covers:
- search_information (query + total_results)
- answer_box (snippet/answer if present)
- knowledge_graph (title + description if present)
- organic_results (top 8: title, link, snippet)
- related_questions (PAA: top 6)
- related_searches (top 8)
Output is capped at 12,000 chars to keep MCP messages tractable.
"""
out: list[str] = []

si = data.get("search_information")
if isinstance(si, dict):
si_parts = []
if "query_displayed" in si:
si_parts.append(f"query_displayed: {si['query_displayed']}")
if "total_results" in si:
si_parts.append(f"total_results: {si['total_results']:,}")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The :, format spec raises ValueError: Cannot specify ',' with 's' if total_results is a string. This checks key presence, but not the value type.

if si_parts:
out.append("=== search_information ===\n" + "\n".join(si_parts))

ab = data.get("answer_box")
if isinstance(ab, dict):
ab_parts = []
if ab.get("type"):
ab_parts.append(f"type: {ab['type']}")
if ab.get("title"):
ab_parts.append(f"title: {ab['title']}")
if ab.get("answer"):
ab_parts.append(f"answer: {ab['answer']}")
if ab.get("snippet"):
ab_parts.append(f"snippet: {ab['snippet']}")
if ab.get("link"):
ab_parts.append(f"link: {ab['link']}")
if ab_parts:
out.append("=== answer_box ===\n" + _truncate("\n ".join(ab_parts), 800))

kg = data.get("knowledge_graph")
if isinstance(kg, dict):
kg_parts = [f"title: {kg.get('title', '?')}", f"type: {kg.get('type', '?')}"]
if kg.get("description"):
kg_parts.append(f"description: {_truncate(kg['description'], 500)}")
out.append("=== knowledge_graph ===\n" + _truncate("\n ".join(kg_parts), 1500))

organic = data.get("organic_results", [])
if isinstance(organic, list) and organic:
lines = [f"organic_results (top {min(len(organic), 8)} of {len(organic)}):"]
for i, r in enumerate(organic[:8], 1):
title = r.get("title", "(no title)")
link = r.get("link", "")
displayed = r.get("displayed_link", "")
snippet = r.get("snippet", "")
lines.append(f"\n [{i}] {title}")
if displayed:
lines.append(f" {displayed}")
elif link:
lines.append(f" {link}")
if snippet:
lines.append(f" snippet: {_truncate(snippet, 300)}")
out.append("\n".join(lines))

paa = data.get("related_questions", [])
if isinstance(paa, list) and paa:
lines = ["related_questions (PAA):"]
for i, q in enumerate(paa[:6], 1):
if not isinstance(q, dict):
continue
question = q.get("question", "?")
snippet = q.get("snippet", "")
title = q.get("title", "")
link = q.get("link", "")
lines.append(f" [{i}] {question}")
if title:
lines.append(f" {title}")
if snippet:
lines.append(f" {_truncate(snippet, 250)}")
if link:
lines.append(f" {link}")
out.append("\n".join(lines))

rs = data.get("related_searches", [])
if isinstance(rs, list) and rs:
qs = [r.get("query", r) if isinstance(r, dict) else r for r in rs[:8]]
out.append("related_searches: " + ", ".join(qs))
Comment on lines +474 to +476

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If an item is a dict without a "query" key, the fallback is the dict itself, and ", ".join(qs) raises TypeError: sequence item N: expected str instance, dict found. Several engines shape related_searches differently - some have block_position, nested items, etc. We needs str(...) coercion or filtering to strings only.


if not out:
# Fall back to JSON if nothing matched (defensive — shouldn't happen
# for any real SerpApi response)
return json.dumps(data, indent=2, ensure_ascii=False)
Comment on lines +478 to +481

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here out will always contain something because search_information is always present. But the problem is that other engines might have results that live under say video_results, local_results, etc. So we might actually have the results for other engines where actual results fields are not extracted.

That was actually the original reasoning not to pursue this path, because we will need to support many diverse engine-output formats that do not always adhere to the same shared interface.

Here, practically though what we can do is for the formatted_output variable (instead of out), we can make it so that it first gets the results first (without including the search information), and if that is empty, then we return full json dump. If it's present, then you can attach the search information and proceed further. Because search information is a metadata, and if we live it as is, then for many engines it will only return the metadata without including any results.

return _truncate("\n\n".join(out), 12_000)


if __name__ == "__main__":
main()
Loading