> SpecStream is the live web data engine for teams and agents. This is the Markdown copy of https://specstream.ai/docs/integrations/dspy. Every page, with what SpecStream does and costs: https://specstream.ai/llms.txt

# 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](https://specstream.ai/integrations/dspy.md)

## 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](https://specstream.ai/docs/account.md#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.  1

    ## 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

    ```bash
    pip install -U dspy requests
    export SPECSTREAM_API_KEY="YOUR_SPECSTREAM_API_KEY"
    ```

2.  2

    ## 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

    ```python
    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.  3

    ## 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

    ```python
    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.  4

    ## 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

    ```python
    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.  5

    ## 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

    ```python
    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](https://specstream.ai/docs/search-api.md).

## 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](https://specstream.ai/docs/billing.md), and send questions to [support@specstream.ai](mailto:support@specstream.ai).

## More

[SpecStream for DSPy](https://specstream.ai/integrations/dspy.md) covers what it does in DSPy on one page. Every other tool has its guide in [Connect your tools](https://specstream.ai/docs/integrations.md). You see the most a request can cost before it starts, and the matches are paid from your balance. [Billing](https://specstream.ai/docs/billing.md) has the details. For help, email [support@specstream.ai](mailto:support@specstream.ai).