DSPy

Give a DSPy program the live web. One function on SpecStream's Search API returns passages for a RAG module or a dspy.ReAct tool, a second reads pages with the Extract API, and the MCP server adds Deep Search lists.

SpecStream for DSPyAgent frameworks & builders

What you need

Python 3.10 or later, with DSPy 3.x.
A SpecStream API key, made under Agents & API in your workspace (more on keys). New accounts get $10 of credit, no card.
For the MCP step: Python 3.12 or later, for the specstream-mcp-stdio bridge.

1. Install DSPy and set your key

The search function below sends the key as a Bearer token. Keep it in the environment, out of your code.

Shell
pip install -U dspy requests
export SPECSTREAM_API_KEY="YOUR_SPECSTREAM_API_KEY"

2. Write the search function

DSPy has no built-in web search: its own tool example returns a placeholder where a search API call would go. This function is that call. It asks the Search API for k results with passages=true, and each of the first k sources then carries an excerpt: the text of its page closest to the query, up to about 3,500 characters. A page not read in time keeps its snippet.

It returns a dspy.Prediction with passages, the shape DSPy's RAG tutorial reads from its retriever, and each passage starts with the page's title and URL so the model can name its source.

specstream_search.py
import os

import dspy
import requests

API = "https://api.specstream.ai"
HEADERS = {"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"}


def search(query: str, k: int = 5) -> dspy.Prediction:
    """Search the live web and return the text of the top pages as passages."""
    response = requests.get(
        f"{API}/v1/web-search",
        params={"q": query, "num": k, "passages": "true"},
        headers=HEADERS,
        timeout=60,
    )
    response.raise_for_status()
    sources = response.json()["data"]["sources"][:k]
    passages = [
        f"{source['title']} ({source['url']})\n{source.get('excerpt') or source.get('snippet', '')}"
        for source in sources
    ]
    return dspy.Prediction(passages=passages)

3. Use it in a RAG module

This is the module from DSPy's RAG tutorial, with the live web as the corpus: forward calls search and passes the passages as context. You can compile it with any DSPy optimizer, as you would with a local index.

rag.py
import dspy

from specstream_search import search

dspy.configure(lm=dspy.LM("openai/gpt-4.1-mini"))


class RAG(dspy.Module):
    def __init__(self, k: int = 5):
        self.k = k
        self.respond = dspy.ChainOfThought("context, question -> response")

    def forward(self, question):
        context = search(question, k=self.k).passages
        return self.respond(context=context, question=question)


rag = RAG()
print(rag(question="What changed in the latest stable release of PostgreSQL?").response)

4. Give a ReAct agent search and page reading

dspy.ReAct takes plain functions as tools and reads their names, type hints and docstrings. With web_search and read_page, the agent finds pages, then reads the ones it needs in full through the Extract API. A page that refuses a plain read is tried again as a browser, then from the site's feed, then from its newest public archive capture.

agent.py
import dspy
import requests

from specstream_search import API, HEADERS, search

dspy.configure(lm=dspy.LM("openai/gpt-4.1-mini"))


def web_search(query: str) -> list[str]:
    """Search the live web. Returns the title, URL and text of the top pages."""
    return search(query).passages


def read_page(url: str) -> str:
    """Read the main text of a web page, PDF, Word or Excel file at a URL, as Markdown."""
    response = requests.post(
        f"{API}/v1/extract",
        headers=HEADERS,
        json={"urls": [url], "format": "markdown"},
        timeout=60,
    )
    response.raise_for_status()
    page = response.json()["data"]["results"][0]
    if page["status"] != "ok":
        return f"Could not read {url}: {page['status']}"
    return page["text"]


agent = dspy.ReAct("question -> answer", tools=[web_search, read_page], max_iters=6)
print(agent(question="Which Python versions does the newest Django release support?").answer)

5. Every SpecStream tool over MCP

Install the bridge with pip install -U "dspy[mcp]" specstream-mcp-stdio. dspy.Tool.from_mcp_tool turns each tool the server lists into a DSPy tool: search_web and read_pages, and the Deep Search tools that return whole lists as structured rows. Keep the agent call inside the open session, and call it with acall.

A Deep Search run starts only after a quote's maximum is approved. Give the key a spending limit under Agents & API, and the agent can never spend above it.

mcp_agent.py
import asyncio
import os

import dspy
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

dspy.configure(lm=dspy.LM("openai/gpt-4.1-mini"))

server = StdioServerParameters(
    command="specstream-mcp-stdio",
    env={
        "SPECSTREAM_API_KEY": os.environ["SPECSTREAM_API_KEY"],
        "SPECSTREAM_MCP_URL": "https://api.specstream.ai/mcp",
    },
)


async def main():
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            response = await session.list_tools()
            tools = [dspy.Tool.from_mcp_tool(session, tool) for tool in response.tools]
            agent = dspy.ReAct("question -> answer", tools=tools, max_iters=8)
            result = await agent.acall(question="What is the latest stable release of PostgreSQL?")
            print(result.answer)


asyncio.run(main())

What the search function reads

Search API field	In each passage
`data.sources[].title`	The first words of the passage
`data.sources[].url`	In brackets after the title, so the model can cite it
`data.sources[].excerpt`	The page's text closest to the query, with `passages=true`
`data.sources[].snippet`	Used when a page was not read in time

Programs that call dspy.Retrieve

Older programs call dspy.Retrieve(k=3) after dspy.configure(rm=...). dspy.Retrieve calls the rm with the query and k, and reads long_text from each item it gets back. Wrap the passages to fit: with from dspy.dsp.utils import dotdict, set dspy.configure(rm=lambda query, k=3, **kwargs: [dotdict(long_text=p) for p in search(query, k=k).passages]).

Searching less while you compile

An optimizer runs your program over the training set many times, and each run that calls search is a charged search. Decorate search with functools.lru_cache while you compile, so each training question is searched once per process.

More results, filters and answers

num takes 1 to 100 results, and page continues the same ranked list. Operators in q hold on every result: site:, -site:, "phrase", filetype: and after:/before:. The same reply also carries answer for a question with one answer, with the page it was read from. Every parameter is in the Search API reference.

If a call fails

raise_for_status raises the HTTP error. 401 means SpecStream got no valid key: check that SPECSTREAM_API_KEY is set where the program runs. A search the engines find nothing for returns 200 with empty sources, so search returns no passages.

Cost

A search is $5 per 1,000 searches of up to 10 results; add speed=fast to the parameters and it costs a fifth of that. A page read is $0.80 per 1,000 pages, with no charge for a page that has no text. Both come from your balance with no quote. New accounts start with $10 of credit, no card. See how billing works, and send questions to [email protected].

SpecStream for DSPy covers what it does in DSPy on one page. Every other tool has its guide in Connect your tools. You see the most a request can cost before it starts, and the matches are paid from your balance. Billing has the details. For help, email [email protected].