Keenable Web Search

uv add "pipecat-ai[keenable]"

import os
from pipecat.services.keenable.search import KeenableWebSearch
from pipecat.processors.aggregators.llm_context import LLMContext

# Keyless (free tier, uses "pro" mode)
search = KeenableWebSearch()

# With API key (uses "realtime" mode by default)
search = KeenableWebSearch(api_key=os.getenv("KEENABLE_API_KEY"))

# Explicit mode selection
search = KeenableWebSearch(
    api_key=os.getenv("KEENABLE_API_KEY"),
    mode="realtime"
)

# Register tools with LLM context
context = LLMContext(tools=await search.tools())

import os
from datetime import date
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.keenable.search import KeenableWebSearch
from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService

# Initialize services
stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
tts = CartesiaTTSService(api_key=os.environ["CARTESIA_API_KEY"])

system_prompt = f"""
You are a helpful assistant with live web access.
Today's date is {date.today():%A, %B %d, %Y}.

You have two tools:
- search_web_pages: search the web for current information
- fetch_page_content: read specific web pages

Use these tools whenever a question needs current information.
"""

llm = OpenAIResponsesLLMService(
    api_key=os.environ["OPENAI_API_KEY"],
    settings=OpenAIResponsesLLMService.Settings(
        system_instruction=system_prompt,
    ),
)

# Initialize web search
search = KeenableWebSearch(api_key=os.getenv("KEENABLE_API_KEY"))

# Create context with tools
context = LLMContext(tools=await search.tools())
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)

# Build pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    user_aggregator,
    llm,
    tts,
    transport.output(),
    assistant_aggregator,
])

worker = PipelineWorker(pipeline)

search = KeenableWebSearch()

# Explicit connection management
await search.start()
try:
    tools = await search.tools()
    # Use tools...
finally:
    await search.close()

# Or use as async context manager
async with KeenableWebSearch() as search:
    tools = await search.tools()
    # Use tools...