feat: add get_weather MCP tool using wttr.in API
- New agent/weather_mcp.py: stdio MCP server with get_weather(location) that returns a short conversational summary (temp, conditions, high/low) - Register weather toolset in build_mcp_toolsets() alongside web-access - Update system prompt with Weather section
This commit is contained in:
+21
-1
@@ -63,12 +63,17 @@ SYSTEM_PROMPT = textwrap.dedent("""\
|
||||
|
||||
# Web access
|
||||
You have web_search and web_scrape tools. Use them when the user asks about
|
||||
current events, recent news, prices, weather, sports scores, or anything
|
||||
current events, recent news, prices, sports scores, or anything
|
||||
that may have changed since your training data. Search first, then scrape
|
||||
a result only if you need more detail. Answer from what you find, in your
|
||||
normal conversational style — don't cite sources formally, just mention the
|
||||
source naturally ("according to..."). If a search comes up empty, say so
|
||||
briefly and move on.
|
||||
|
||||
# Weather
|
||||
You can check the weather for any location. Use get_weather when the user
|
||||
asks about current conditions, temperature, or forecasts. It returns a
|
||||
short summary — relay it naturally in your own words.
|
||||
""")
|
||||
|
||||
|
||||
@@ -178,6 +183,21 @@ def build_mcp_toolsets() -> list[mcp.MCPToolset]:
|
||||
)
|
||||
logger.info("Web-access MCP toolset enabled (Firecrawl at %s)", FIRECRAWL_BASE)
|
||||
|
||||
python_bin = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".venv", "bin", "python")
|
||||
weather_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weather_mcp.py")
|
||||
toolsets.append(
|
||||
mcp.MCPToolset(
|
||||
id="weather",
|
||||
mcp_server=mcp.MCPServerStdio(
|
||||
command=python_bin,
|
||||
args=[weather_mcp_script],
|
||||
env={**os.environ},
|
||||
client_session_timeout_seconds=30,
|
||||
),
|
||||
)
|
||||
)
|
||||
logger.info("Weather MCP toolset enabled (wttr.in)")
|
||||
|
||||
extra = os.environ.get("EXTRA_MCP_SERVERS", "")
|
||||
if extra:
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
Metadata-Version: 2.4
|
||||
Name: voice-agent
|
||||
Version: 0.1.0
|
||||
Summary: Real-time voice assistant: Azure STT/TTS + Gemma LLM via LiveKit Agents
|
||||
Requires-Python: <3.15,>=3.10
|
||||
Requires-Dist: livekit-agents[mcp]~=1.7
|
||||
Requires-Dist: livekit-plugins-azure~=1.7
|
||||
Requires-Dist: livekit-plugins-openai~=1.7
|
||||
Requires-Dist: python-dotenv
|
||||
@@ -0,0 +1,6 @@
|
||||
pyproject.toml
|
||||
voice_agent.egg-info/PKG-INFO
|
||||
voice_agent.egg-info/SOURCES.txt
|
||||
voice_agent.egg-info/dependency_links.txt
|
||||
voice_agent.egg-info/requires.txt
|
||||
voice_agent.egg-info/top_level.txt
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
livekit-agents[mcp]~=1.7
|
||||
livekit-plugins-azure~=1.7
|
||||
livekit-plugins-openai~=1.7
|
||||
python-dotenv
|
||||
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Weather MCP server — exposes a get_weather tool backed by wttr.in.
|
||||
|
||||
Runs over stdio inside the voice container. The agent attaches it via
|
||||
MCPServerStdio, so the LLM can call get_weather during a conversation to
|
||||
report current conditions and today's forecast for any location.
|
||||
|
||||
Backed by wttr.in (no API key needed):
|
||||
- GET https://wttr.in/{location}?format=j1 -> full JSON weather report
|
||||
|
||||
The tool returns a short, conversational summary suitable for speech
|
||||
(1-2 sentences), not raw data.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import httpx
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
WTTR_BASE = os.environ.get("WTTR_BASE", "https://wttr.in").rstrip("/")
|
||||
|
||||
mcp = FastMCP("weather")
|
||||
|
||||
|
||||
def _summarize(data: dict, location: str) -> str:
|
||||
"""Build a short spoken-style weather summary from wttr.in j1 JSON."""
|
||||
current = data.get("current_condition", [{}])[0] or {}
|
||||
area = data.get("nearest_area", [{}])[0] or {}
|
||||
|
||||
city = (area.get("areaName", [{}])[0].get("value") or location).strip()
|
||||
temp_c = current.get("temp_C")
|
||||
feels_c = current.get("FeelsLikeC")
|
||||
desc = (current.get("weatherDesc", [{}])[0].get("value") or "").strip().lower()
|
||||
|
||||
astronomy = data.get("astronomy", [{}])[0] or {}
|
||||
sunrise = astronomy.get("sunrise", "")
|
||||
sunset = astronomy.get("sunset", "")
|
||||
|
||||
today = (data.get("weather") or [None])[0] or {}
|
||||
high_c = today.get("maxtempC")
|
||||
min_c = today.get("mintempC")
|
||||
|
||||
parts: list[str] = []
|
||||
|
||||
if temp_c is not None:
|
||||
lead = f"It's {temp_c} degrees and {desc or 'clear'} in {city}"
|
||||
if feels_c is not None and feels_c != temp_c:
|
||||
lead += f", feels like {feels_c}"
|
||||
parts.append(lead + ".")
|
||||
|
||||
if high_c is not None and min_c is not None:
|
||||
parts.append(f"Today's high is {high_c} and the low is {min_c}.")
|
||||
|
||||
if sunrise and sunset:
|
||||
parts.append(f"Sunrise was at {sunrise} and sunset is at {sunset}.")
|
||||
|
||||
return " ".join(parts) or f"I have no detailed weather for {city} right now."
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def get_weather(location: str) -> str:
|
||||
"""Get the current weather and today's forecast for a location.
|
||||
|
||||
Use this when the user asks about current conditions, temperature,
|
||||
or the forecast for a place. Returns a short conversational summary
|
||||
suitable for reading aloud (1-2 sentences).
|
||||
"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
resp = await client.get(
|
||||
f"{WTTR_BASE}/{location}?format=j1",
|
||||
headers={"User-Agent": "curl/8.0"},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
except Exception: # noqa: BLE001
|
||||
return "I couldn't get the weather for that location"
|
||||
|
||||
try:
|
||||
summary = _summarize(data, location)
|
||||
except Exception: # noqa: BLE001
|
||||
return "I couldn't get the weather for that location"
|
||||
return summary
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run(transport="stdio")
|
||||
Reference in New Issue
Block a user