Web Tools

# default — GR's backdated corpus
(nothing to set)

# or route the same tools to Tavily
export OPENREWARD_SEARCH_BACKEND=tavily
export TAVILY_API_KEY="tvly-xxxxxxxx"

from openreward.environments import Environment, Split, TextBlock
from openreward.toolsets import WebToolset

class ResearchEnv(Environment):
    toolsets = [WebToolset]

    @classmethod
    def list_splits(cls):
        return [Split(name="test", type="test")]

    @classmethod
    def list_tasks(cls, split):
        return [{"id": "1", "question": "What happened with X in late 2025?"}]

    def get_prompt(self):
        return [TextBlock(text="Research the question using web_search and web_fetch, then answer with sources.")]

from openreward import OpenReward

or_client = OpenReward()
env = or_client.environments.get(name="MyOrg/ResearchEnv")
tasks = env.list_tasks(split="test")

with env.session(task=tasks[0], toolset="web") as session:
    result = session.call_tool("web_search", {"query": "US interest rate decision December 2025"})
    print(result.blocks[0].text)

class ResearchEnv(Environment):
    toolsets = [WebToolset]

    def __init__(self, task_spec, secrets={}):
        super().__init__(task_spec, secrets)
        self.search_secrets = secrets

with env.session(
    task=task,
    secrets={"tavily_api_key": "tvly-xxxxxxxx"},
) as session:
    ...

class ResearchEnv(Environment):
    toolsets = [WebToolset]
    search_backend = "tavily"      # attribute, @property, or zero-arg callable

from openreward.toolsets import WebToolset, CLIToolset

class ResearchAndCodeEnv(Environment):
    toolsets = [WebToolset, CLIToolset]

import asyncio
from openreward.tools import Search, Fetch

async def main():
    s = Search()                       # OPENREWARD_SEARCH_BACKEND picks the provider
    result = await s.run("US interest rate decision December 2025")
    if result.ok:
        print(result.output)               # citation-ready text
        print(result.data["mode"])         # which backend answered
        for hit in result.data["hits"]:
            print(hit["title"], "->", hit["url"])

    f = Fetch(backend="tavily")        # or pin one explicitly
    page = await f.run("https://example.com/article", "Summarise the key points")
    if page.ok:
        print(page.data["content"])

asyncio.run(main())

from openreward.search_backends import SearchBackend, register_backend
from openreward.tools.web import WebToolResult, format_search_output, validate_search_input

class ExaBackend(SearchBackend):
    name = "exa"
    supports_as_of = False

    def __init__(self, *, config=None, secrets=None):
        ...

    async def search(self, *, query, k=None, as_of=None,
                     allowed_domains=None, blocked_domains=None) -> WebToolResult:
        query, allowed, blocked, error = validate_search_input(query, allowed_domains, blocked_domains)
        if error is not None:
            return error
        hits = [...]                                   # {title, url, snippet, score}
        return WebToolResult.success(
            format_search_output(query, hits),         # shared renderer — do not hand-roll
            {"query": query, "hits": hits, "mode": "exa"},
        )

    async def fetch(self, *, url, prompt, as_of=None) -> WebToolResult:
        ...

register_backend("exa", ExaBackend)