Ground Fireworks AI Models in Real-Time Web Data with You.com

pip install --quiet --upgrade openai
export FIREWORKS_API_KEY="<fireworks-api-key>"

import os
from openai import OpenAI

fireworks = OpenAI(
    base_url="https://api.fireworks.ai/inference/v1",
    api_key=os.environ["FIREWORKS_API_KEY"],
)

response = fireworks.responses.create(
    model="accounts/fireworks/models/kimi-k3",
    input="What are the most recent developments in fusion energy? Cite your sources.",
    tools=[
        {
            "type": "mcp",
            "server_label": "you",
            "server_url": "https://api.you.com/mcp?profile=free",
            "require_approval": "never",
        }
    ],
)
print(response.output_text)

pip install --quiet --upgrade openai requests

export YDC_API_KEY="<you-api-key>"
export FIREWORKS_API_KEY="<fireworks-api-key>"

import json
import os

import requests
from openai import OpenAI

fireworks = OpenAI(
    base_url="https://api.fireworks.ai/inference/v1",
    api_key=os.environ["FIREWORKS_API_KEY"],
)

CHAT_MODEL = "accounts/fireworks/models/gpt-oss-120b"
AGENT_MODEL = "accounts/fireworks/models/kimi-k3"

def web_search(query, count=5):
    response = requests.post(
        "https://ydc-index.io/v1/search",
        headers={"X-API-Key": os.environ["YDC_API_KEY"]},
        json={"query": query, "count": count},
        timeout=30,
    )
    response.raise_for_status()
    return [
        {"title": r["title"], "url": r["url"], "snippets": r.get("snippets", [])}
        for r in response.json()["results"].get("web", [])
    ]

for result in web_search("fusion energy breakthroughs"):
    print(f'{result["title"]}\n  {result["url"]}')

CITATION_PROMPT = (
    "Answer the question using only the numbered search results. Cite every "
    "claim inline with the number of its source, such as [1]. If the results "
    "don't contain the answer, say so."
)

def answer(question):
    sources = web_search(question)
    context = "\n\n".join(
        f'[{i}] {s["title"]} ({s["url"]})\n' + "\n".join(s["snippets"])
        for i, s in enumerate(sources, 1)
    )
    response = fireworks.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": CITATION_PROMPT},
            {"role": "user", "content": f"Search results:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return response.choices[0].message.content, sources

text, sources = answer("What are the latest milestones in fusion energy?")
print(text)
for i, source in enumerate(sources, 1):
    print(f'[{i}] {source["url"]}')

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "Search the web for current information. Returns "
            "result titles, URLs, and relevant snippets.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "A short search query about one topic.",
                    }
                },
                "required": ["query"],
            },
        },
    }
]

AGENT_PROMPT = (
    "You are a research assistant. Use web_search to find current "
    "information, and run separate searches for separate topics. Cite "
    "sources inline with their URLs."
)

def run_agent(question, max_steps=8):
    messages = [
        {"role": "system", "content": AGENT_PROMPT},
        {"role": "user", "content": question},
    ]
    for _ in range(max_steps):
        response = fireworks.chat.completions.create(
            model=AGENT_MODEL, messages=messages, tools=TOOLS
        )
        message = response.choices[0].message
        messages.append(message.model_dump(exclude_none=True))

        if not message.tool_calls:
            return message.content

        for call in message.tool_calls:
            args = json.loads(call.function.arguments)
            print(f'Searching: {args["query"]}')
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": json.dumps(web_search(**args)),
                }
            )

    raise RuntimeError("The agent didn't finish within max_steps.")

print(run_agent("Compare the latest funding rounds of two fusion energy startups."))

pip install --quiet --upgrade langchain langchain-youdotcom langchain-fireworks

from langchain.agents import create_agent
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_fireworks import ChatFireworks
from langchain_youdotcom import YouRetriever, YouSearchTool

retriever = YouRetriever(count=5)
llm = ChatFireworks(model="accounts/fireworks/models/gpt-oss-120b")

prompt = ChatPromptTemplate.from_template(
    "Answer using only these search results, and cite their URLs.\n\n"
    "{context}\n\nQuestion: {question}"
)

def format_docs(docs):
    return "\n\n".join(f'{doc.metadata["url"]}\n{doc.page_content}' for doc in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
print(chain.invoke("What are the latest milestones in fusion energy?"))

agent = create_agent(
    model=ChatFireworks(model="accounts/fireworks/models/kimi-k3"),
    tools=[YouSearchTool()],
    system_prompt="You are a research assistant. Search the web, and cite sources with their URLs.",
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Who leads the fusion energy race, and why?"}]}
)
print(result["messages"][-1].content)