perception-tools¶
第4章 · 工具 · 配套项目
chapter4/perception-tools
项目说明¶
Perception Tools MCP Server¶
A comprehensive MCP (Model Context Protocol) server providing various perception and data retrieval capabilities for AI agents.
Features¶
✨ No API Keys Required! Most features work out-of-the-box with free, open APIs.
🔍 Search Tools¶
- Web Search: DuckDuckGo search (free, no API key required)
- Knowledge Base Search: Search local document collections
- File Download: Download files from URLs with safety checks
📄 Multimodal Understanding Tools¶
- Web Page Reader: Extract text and links from web pages
- Document Reader: Extract content from PDF, DOCX, PPTX files
- Image Parser: Parse and analyze image files
- Video Parser: Extract metadata from video files
📁 File System Tools¶
- File Reader: Read files with encoding support
- Grep Search: Search for patterns in files (regex support)
- Text Summarization: Summarize long text content
🌐 Public Data Sources¶
- Weather: Current weather via Open-Meteo (free, no API key)
- Stock Prices: Real-time stock data from Yahoo Finance (free, no API key)
- Crypto Prices: Cryptocurrency prices via CoinGecko (free, no API key)
- Currency Conversion: Convert between currencies (free, no API key)
- Location Search: Geocoding via Nominatim (OpenStreetMap) (free, no API key)
- POI Search: Points of Interest via Overpass API (OpenStreetMap) (free, no API key)
- Wikipedia: Search and retrieve Wikipedia articles (free, no API key)
- ArXiv: Search academic papers on ArXiv (free, no API key)
- Wayback Machine: Access archived web pages (free, no API key)
🔐 Private Data Sources¶
- Google Calendar: Query calendar events
- Notion: Search Notion workspace
Installation¶
- Clone the repository and navigate to the project directory:
- Install dependencies:
- No additional configuration required! The server works out-of-the-box with free APIs.
Configuration¶
✅ Default Free APIs (No Setup Required)¶
The following features work immediately without any API keys:
- Web Search: DuckDuckGo
- Weather: Open-Meteo
- Stock Prices: Yahoo Finance
- Crypto Prices: CoinGecko
- Currency Conversion: ExchangeRate-API
- Location Search: Nominatim (OpenStreetMap)
- POI Search: Overpass API (OpenStreetMap)
- Wikipedia: Wikipedia API
- ArXiv: ArXiv API
- Wayback Machine: Internet Archive
Optional Private Data Integrations¶
Google Calendar¶
For Google Calendar integration, you need to set up OAuth2:
Follow the Google Calendar API quickstart to set up OAuth2 credentials.
Notion¶
- Create a Notion integration at notion.so/my-integrations
- Get your integration token
- Share your databases/pages with the integration
- Add
NOTION_API_KEYto.env
Usage¶
Running the MCP Server¶
The server runs using stdio transport, suitable for integration with MCP clients.
Command-Line Interface (cli.py)¶
除了以 MCP stdio 协议对外服务，仓库根目录提供了一个统一的命令行入口
cli.py，无需 MCP 客户端即可直接列出、查看、调用和演示各类感知工具。
工具按第四章「感知工具」的五类场景组织：搜索 / 多模态理解 / 文件系统 /
公开数据源 / 私有数据源（当前共 53 个工具）。
# 查看帮助（中文）
python cli.py --help
# 按五类列出全部感知工具（可用 --category 只看某一类）
python cli.py list
python cli.py list --category filesystem
# 查看某个工具的参数签名与调用示例
python cli.py info weather
# 直接调用某个工具，参数以 key=value 形式传入，结果为标准 ActionResponse JSON
python cli.py run grep 'pattern=async def' directory=src 'file_pattern=*.py'
python cli.py run currency_converter amount=100 from_currency=USD to_currency=CNY
# 运行端到端演示：串联「本地资料 + 外部信息」的研究助手 Agent 感知流程
python cli.py demo # 完整演示（含联网步骤）
python cli.py demo --offline # 离线演示（只跑文件系统 / 本地知识库等不联网步骤）
说明：
- 每个工具都是异步函数，返回统一的
ActionResponse（JSON）；CLI 负责运行事件 循环、解析 JSON 并友好打印。 - 工具按需惰性导入：
list/info/ 离线demo在缺少可选依赖（如whisper、waybackpy）时仍可正常工作，只有真正调用相关工具时才导入对应模块。 - 需要联网的工具在
list中标注「联网」，需要授权/API Key 的工具标注了对应说明。
Using with MCP Clients¶
Configure your MCP client (e.g., Claude Desktop) to connect to this server:
{
"mcpServers": {
"perception-tools": {
"command": "python",
"args": ["/path/to/perception-tools/src/main.py"]
}
}
}
Available Tools¶
Search Tools¶
web_search¶
Search the web using DuckDuckGo (free, no API key required).
Parameters:
- query (str): Search query string
- num_results (int, default=5): Number of results (1-10)
- region (str, default="wt-wt"): Region code (e.g., "us-en", "uk-en", "wt-wt" for worldwide)
download¶
Download a file from a URL.
Parameters:
- url (str): URL to download from
- output_path (str): Local path to save the file
- overwrite (bool, default=False): Overwrite existing file
- timeout (int, default=180): Download timeout in seconds
knowledge_base_search¶
Search a local knowledge base directory.
Parameters:
- query (str): Search query
- knowledge_base_path (str): Path to knowledge base directory
- top_k (int, default=5): Number of top results
Multimodal Understanding Tools¶
webpage_reader¶
Read and extract content from a webpage.
Parameters:
- url (str): URL of the webpage
- extract_text (bool, default=True): Extract text content
- extract_links (bool, default=False): Extract links
document_reader¶
Read and extract content from documents (PDF, DOCX, PPTX).
Parameters:
- file_path (str): Path to document file or URL
- extract_images (bool, default=False): Extract images
image_parser¶
Parse and analyze image files.
Parameters:
- image_path (str): Path to image file or URL
- use_llm (bool, default=True): Use LLM for analysis
Vision LLM keys / OpenRouter fallback: AI image/video analysis (
analyze_image_ai/analyze_video_ai) useOPENAI_API_KEYwhen set. If it is absent butOPENROUTER_API_KEYis set, they transparently route through OpenRouter (base_url=https://openrouter.ai/api/v1, model mapped toprovider/modelform). Override the model viaPERCEPTION_VISION_MODEL. (Local Whisper transcription still needsOPENAI_API_KEY— OpenRouter has no audio-transcription API.)
video_parser¶
Parse and extract metadata from video files.
Parameters:
- video_path (str): Path to video file or URL
- extract_frames (bool, default=False): Extract sample frames
- frame_interval (int, default=30): Frame extraction interval
File System Tools¶
file_reader¶
Read a file and return its contents.
Parameters:
- file_path (str): Path to the file
- encoding (str, default="utf-8"): File encoding
- max_length (int, default=50000): Maximum characters to read
grep¶
Search for patterns in files (grep-like functionality).
Parameters:
- pattern (str): Regular expression pattern
- directory (str): Directory to search in
- file_pattern (str, default=""): File pattern (e.g., .py)
- recursive (bool, default=True): Search recursively
- case_sensitive (bool, default=False): Case-sensitive search
- max_results (int, default=100): Maximum results
text_summarizer¶
Summarize long text content.
Parameters:
- text (str): Text to summarize
- max_length (int, default=500): Target summary length
- use_llm (bool, default=True): Use LLM for summarization
Public Data Source Tools¶
weather¶
Get current weather information using Open-Meteo API (free, no API key required).
Parameters:
- location (str): City name (automatically geocoded)
- latitude (float, optional): Latitude coordinate
- longitude (float, optional): Longitude coordinate
stock_price¶
Get stock price and market information using Yahoo Finance (free, no API key required).
Parameters:
- symbol (str): Stock ticker symbol (e.g., AAPL, TSLA, GOOGL)
- interval (str, default="1d"): Data interval
crypto_price¶
Get cryptocurrency price information using CoinGecko API (free, no API key required).
Parameters:
- symbol (str): Cryptocurrency symbol or ID (e.g., bitcoin, ethereum, btc, eth)
- vs_currency (str, default="usd"): Target currency (usd, eur, gbp, etc.)
currency_converter¶
Convert between currencies.
Parameters:
- amount (float): Amount to convert
- from_currency (str): Source currency code (e.g., USD)
- to_currency (str): Target currency code (e.g., EUR)
wikipedia_search¶
Search Wikipedia and get article summary.
Parameters:
- query (str): Search query
- language (str, default="en"): Wikipedia language
- sentences (int, default=5): Summary sentence count
arxiv_search¶
Search ArXiv for academic papers.
Parameters:
- query (str): Search query
- max_results (int, default=5): Maximum results
- sort_by (str, default="relevance"): Sort method
wayback_search¶
Search Wayback Machine for archived web pages.
Parameters:
- url (str): URL to search for
- year (int, optional): Filter by year
- limit (int, default=10): Maximum snapshots
location_search¶
Search for locations using Nominatim (OpenStreetMap) API (free, no API key required).
Parameters:
- query (str): Location query (e.g., "Eiffel Tower", "New York", "Tokyo")
- limit (int, default=5): Maximum number of results (1-50)
- country_code (str, optional): Country code filter (e.g., "us", "gb", "fr")
poi_search¶
Search for Points of Interest near a location using Overpass API (free, no API key required).
Parameters:
- query (str): Type of POI (e.g., "restaurant", "cafe", "hospital", "atm", "hotel")
- latitude (float): Center latitude coordinate
- longitude (float): Center longitude coordinate
- radius (int, default=1000): Search radius in meters
- limit (int, default=10): Maximum number of results
Private Data Source Tools¶
calendar_events¶
Get events from Google Calendar.
Parameters:
- start_date (str, optional): Start date (ISO format)
- end_date (str, optional): End date (ISO format)
- calendar_id (str, default="primary"): Calendar ID
- max_results (int, default=10): Maximum events
notion_search¶
Search Notion workspace.
Parameters:
- query (str): Search query
- database_id (str, optional): Specific database ID
- page_size (int, default=10): Results per page
Architecture¶
The project follows SOLID principles with a modular architecture:
perception-tools/
├── src/
│ ├── main.py # MCP server entry point
│ ├── base.py # Base models and utilities
│ ├── search_tools.py # Search functionality
│ ├── multimodal_tools.py # Document/media processing
│ ├── filesystem_tools.py # File operations
│ ├── public_data_tools.py # Public APIs
│ └── private_data_tools.py # Private data sources
├── requirements.txt # Python dependencies
├── env.example # Environment variables template
└── README.md # This file
Error Handling¶
All tools return a standardized ActionResponse format:
{
"success": true/false,
"message": "Result data or error message",
"metadata": {
"additional": "context information"
}
}
Contributing¶
Contributions are welcome! Please ensure: 1. Code follows KISS, DRY, and SOLID principles 2. All tools return standardized ActionResponse format 3. Proper error handling and logging 4. Documentation for new tools
License¶
This project is part of the AI Agent training camp materials.
Support¶
For issues and questions, please refer to the main project documentation.
源代码¶
cli.py¶
#!/usr/bin/env python3
"""
感知工具 MCP 服务器 —— 统一命令行入口（实验 4-1）。
除了以 MCP stdio 协议对外提供服务（见 src/main.py），本文件提供一个不依赖
MCP 客户端的命令行入口，方便直接列出、调用和演示各类感知工具：
python cli.py list # 按五大类列出全部感知工具
python cli.py info <tool> # 查看某个工具的参数签名
python cli.py run <tool> k=v ... # 直接调用某个工具并打印 JSON 结果
python cli.py demo [--offline] # 运行一个端到端的感知场景演示
工具按《深入理解 AI Agent》第四章「感知工具」的五类场景组织：
搜索、多模态理解、文件系统、公开数据源、私有数据源。
设计说明：
- 每个工具都是异步函数，返回统一的 ActionResponse（JSON）。CLI 负责运行事件
循环、解析 JSON 并友好打印。
- 工具按需惰性导入：只有真正调用某个工具时才导入其所在模块，因此在缺少
可选依赖（如 yfinance、opencv、whisper）时，list / info / 离线 demo 仍可正常工作。
"""
import argparse
import asyncio
import importlib
import inspect
import json
import logging
import sys
import tempfile
import typing
from pathlib import Path
SRC_DIR = Path(__file__).parent / "src"
sys.path.insert(0, str(SRC_DIR))
# 五大类的中文标题（与书中实验 4-1 的分类一一对应）
CATEGORIES = {
"search": "搜索",
"multimodal": "多模态理解",
"filesystem": "文件系统",
"public": "公开数据源",
"private": "私有数据源",
}
class Tool(typing.NamedTuple):
"""一个感知工具的注册项。"""
name: str # CLI / MCP 中暴露的工具名
category: str # 所属分类（CATEGORIES 的 key）
module: str # src/ 下的模块名
func: str # 模块中的异步函数名
desc: str # 一句话中文描述
online: bool = False # 是否需要联网
note: str = "" # 额外说明（如需要 API Key / 授权）
# ---------------------------------------------------------------------------
# 工具注册表：与 src/main.py 暴露的 MCP 工具保持一致，并补齐 README 中已声明、
# 但此前未在 main.py 注册的三个工具（crypto_price / location_search / poi_search）。
# ---------------------------------------------------------------------------
TOOLS: list[Tool] = [
# ---- 搜索 ----
Tool("web_search", "search", "search_tools", "search_web",
"使用 DuckDuckGo 进行网络搜索（免费，无需 API Key）", online=True),
Tool("knowledge_base_search", "search", "search_tools", "search_knowledge_base",
"在本地知识库目录中做全文检索"),
Tool("download", "search", "search_tools", "download_file",
"从 URL 下载文件到本地（含大小/覆盖保护）", online=True),
Tool("google_search_enhanced", "search", "google_search_enhanced", "google_search_api",
"Google Custom Search，失败时回退 DuckDuckGo", online=True,
note="Google API 需 GOOGLE_API_KEY，未配置则自动回退"),
# ---- 多模态理解 ----
Tool("webpage_reader", "multimodal", "multimodal_tools", "read_webpage",
"抓取并提取网页正文/链接", online=True),
Tool("webpage_read_enhanced", "multimodal", "google_search_enhanced", "read_webpage_content",
"增强版网页正文提取", online=True),
Tool("document_reader", "multimodal", "multimodal_tools", "read_document",
"读取 PDF/DOCX/PPTX 文档内容"),
Tool("pdf_extract", "multimodal", "document_processing_tools", "extract_pdf_text",
"提取 PDF 文本（支持页码范围）"),
Tool("docx_extract", "multimodal", "document_processing_tools", "extract_docx_content",
"提取 Word（DOCX）文档内容"),
Tool("pptx_extract", "multimodal", "document_processing_tools", "extract_pptx_content",
"提取 PowerPoint（PPTX）内容"),
Tool("csv_parse", "multimodal", "document_processing_tools", "extract_csv_content",
"解析 CSV 表格数据"),
Tool("image_parser", "multimodal", "multimodal_tools", "parse_image",
"解析图片（可选 LLM 视觉分析）", note="use_llm 需视觉模型 API"),
Tool("image_ocr", "multimodal", "media_processing_tools", "extract_text_ocr",
"对图片做 OCR 文字识别", note="需安装 tesseract"),
Tool("image_analyze", "multimodal", "media_processing_tools", "analyze_image_ai",
"用视觉模型分析图片内容", note="需视觉模型 API"),
Tool("image_metadata", "multimodal", "media_processing_tools", "get_image_metadata",
"读取图片 EXIF 等元数据"),
Tool("video_parser", "multimodal", "multimodal_tools", "parse_video",
"提取视频元数据/采样帧"),
Tool("video_keyframes", "multimodal", "media_processing_tools", "extract_video_keyframes",
"从视频抽取关键帧"),
Tool("video_analyze", "multimodal", "media_processing_tools", "analyze_video_ai",
"用视觉模型分析视频内容", note="需视觉模型 API"),
Tool("audio_transcribe", "multimodal", "media_processing_tools", "transcribe_audio_whisper",
"用 Whisper 将音频转写为文本", note="需安装 whisper"),
Tool("audio_metadata", "multimodal", "media_processing_tools", "extract_audio_metadata",
"读取音频文件元数据"),
Tool("audio_trim", "multimodal", "media_processing_tools", "trim_audio",
"裁剪音频到指定时间区间"),
Tool("youtube_transcript", "multimodal", "multimodal_tools", "extract_youtube_transcript",
"提取 YouTube 视频字幕", online=True),
Tool("youtube_download", "multimodal", "multimodal_tools", "download_youtube_video",
"下载 YouTube 视频", online=True),
# ---- 文件系统 ----
Tool("file_reader", "filesystem", "filesystem_tools", "read_file",
"读取文件内容（支持编码与截断）"),
Tool("grep", "filesystem", "filesystem_tools", "grep_search",
"在目录中按正则搜索文件内容"),
Tool("text_summarizer", "filesystem", "filesystem_tools", "summarize_text",
"对长文本做摘要（抽取式/截断，占位实现）"),
# ---- 公开数据源 ----
Tool("weather", "public", "public_data_tools", "get_weather",
"查询天气（Open-Meteo，免费无 Key）", online=True),
Tool("stock_price", "public", "public_data_tools", "get_stock_price",
"查询股票行情", online=True),
Tool("crypto_price", "public", "public_data_tools", "get_crypto_price",
"查询加密货币价格（CoinGecko，免费无 Key）", online=True),
Tool("currency_converter", "public", "public_data_tools", "convert_currency",
"货币汇率换算（免费无 Key）", online=True),
Tool("wikipedia_search", "public", "public_data_tools", "search_wikipedia",
"搜索 Wikipedia 并返回摘要", online=True),
Tool("arxiv_search", "public", "public_data_tools", "search_arxiv",
"搜索 ArXiv 学术论文", online=True),
Tool("wayback_search", "public", "public_data_tools", "search_wayback",
"在 Wayback Machine 查历史快照", online=True),
Tool("location_search", "public", "public_data_tools", "search_location",
"地名/地点地理编码（Nominatim，免费无 Key）", online=True),
Tool("poi_search", "public", "public_data_tools", "search_poi",
"查询坐标附近的兴趣点（Overpass，免费无 Key）", online=True),
Tool("yfinance_quote", "public", "yahoo_finance_tools", "get_stock_quote",
"Yahoo Finance 实时报价", online=True),
Tool("yfinance_historical", "public", "yahoo_finance_tools", "get_historical_data",
"Yahoo Finance 历史行情", online=True),
Tool("yfinance_company_info", "public", "yahoo_finance_tools", "get_company_info",
"Yahoo Finance 公司资料", online=True),
Tool("yfinance_financials", "public", "yahoo_finance_tools", "get_financial_statements",
"Yahoo Finance 财务报表", online=True),
Tool("pubchem_search", "public", "pubchem_tools", "search_compounds",
"在 PubChem 搜索化合物", online=True),
Tool("pubchem_properties", "public", "pubchem_tools", "get_compound_properties",
"获取 PubChem 化合物属性", online=True),
Tool("pubchem_synonyms", "public", "pubchem_tools", "get_compound_synonyms",
"获取 PubChem 化合物别名", online=True),
Tool("pubchem_similar", "public", "pubchem_tools", "search_similar_compounds",
"搜索结构相似的化合物", online=True),
Tool("wiki_article_full", "public", "wiki_enhanced", "get_article_content",
"获取 Wikipedia 条目全文", online=True),
Tool("wiki_article_categories", "public", "wiki_enhanced", "get_article_categories",
"获取 Wikipedia 条目分类", online=True),
Tool("wiki_article_links", "public", "wiki_enhanced", "get_article_links",
"获取 Wikipedia 条目中的链接", online=True),
Tool("wiki_article_history", "public", "wiki_enhanced", "get_article_history",
"获取 Wikipedia 条目历史版本", online=True),
Tool("arxiv_paper_details", "public", "arxiv_enhanced", "get_paper_details",
"获取 ArXiv 论文详情", online=True),
Tool("arxiv_download", "public", "arxiv_enhanced", "download_paper",
"下载 ArXiv 论文 PDF", online=True),
Tool("arxiv_categories", "public", "arxiv_enhanced", "get_arxiv_categories",
"列出 ArXiv 学科分类", online=True),
Tool("wayback_archived_content", "public", "wayback_enhanced", "get_archived_content",
"获取 Wayback 存档页面内容", online=True),
# ---- 私有数据源 ----
Tool("calendar_events", "private", "private_data_tools", "get_calendar_events",
"读取 Google 日历事件", online=True, note="需 Google OAuth 授权"),
Tool("notion_search", "private", "private_data_tools", "search_notion",
"搜索 Notion 工作区", online=True, note="需 NOTION_API_KEY"),
]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}
# ---------------------------------------------------------------------------
# 调用辅助
# ---------------------------------------------------------------------------
def _load_callable(tool: Tool):
"""惰性导入并返回工具对应的异步函数。"""
module = importlib.import_module(tool.module)
return getattr(module, tool.func)
def _coerce(value: str, annotation):
"""把命令行传入的字符串按函数注解转换成合适的类型。"""
# 解开 Optional[X] / X | None
origin = typing.get_origin(annotation)
if origin is typing.Union or (origin is not None and str(origin) == "<class 'types.UnionType'>"):
args = [a for a in typing.get_args(annotation) if a is not type(None)]
annotation = args[0] if args else str
origin = typing.get_origin(annotation)
if annotation is bool:
return value.strip().lower() in ("1", "true", "yes", "y", "on")
if annotation is int:
return int(value)
if annotation is float:
return float(value)
if annotation in (list, dict) or origin in (list, dict):
return json.loads(value)
return value
def _parse_kwargs(func, pairs: list[str]) -> dict:
"""把 key=value 列表解析成传给工具函数的关键字参数。"""
sig = inspect.signature(func)
kwargs = {}
for pair in pairs:
if "=" not in pair:
raise ValueError(f"参数必须是 key=value 形式：{pair!r}")
key, _, raw = pair.partition("=")
key = key.strip()
if key not in sig.parameters:
valid = ", ".join(sig.parameters)
raise ValueError(f"未知参数 {key!r}，可用参数：{valid}")
kwargs[key] = _coerce(raw, sig.parameters[key].annotation)
return kwargs
def _unwrap(result):
"""工具返回 TextContent(JSON) 或裸 JSON 字符串，统一解析成 dict。"""
text = getattr(result, "text", result)
if isinstance(text, (dict, list)):
return text
try:
return json.loads(text)
except (json.JSONDecodeError, TypeError):
return {"success": True, "message": text, "metadata": {}}
async def _invoke(tool: Tool, kwargs: dict) -> dict:
func = _load_callable(tool)
result = await func(**kwargs)
return _unwrap(result)
# ---------------------------------------------------------------------------
# 子命令实现
# ---------------------------------------------------------------------------
def cmd_list(args) -> int:
print("\n感知工具 MCP 服务器 —— 工具清单（共 {} 个）".format(len(TOOLS)))
print("=" * 72)
cats = [args.category] if args.category else list(CATEGORIES)
for cat in cats:
tools = [t for t in TOOLS if t.category == cat]
if not tools:
continue
print(f"\n【{CATEGORIES[cat]}】({len(tools)} 个)")
print("-" * 72)
for t in tools:
flags = []
if t.online:
flags.append("联网")
if t.note:
flags.append(t.note)
tag = f" [{'；'.join(flags)}]" if flags else ""
print(f" {t.name:<26} {t.desc}{tag}")
print("\n提示：`python cli.py info <tool>` 查看参数；`python cli.py run <tool> k=v` 调用。\n")
return 0
def cmd_info(args) -> int:
tool = TOOLS_BY_NAME.get(args.tool)
if tool is None:
print(f"未找到工具：{args.tool}。用 `python cli.py list` 查看全部。", file=sys.stderr)
return 1
try:
func = _load_callable(tool)
except Exception as e:
print(f"工具 {tool.name} 所在模块导入失败（可能缺少可选依赖）：{e}", file=sys.stderr)
return 1
sig = inspect.signature(func)
print(f"\n工具：{tool.name} 分类：{CATEGORIES[tool.category]}")
print(f"描述：{tool.desc}")
print(f"实现：src/{tool.module}.py :: {tool.func}()")
if tool.online:
print("需要联网：是")
if tool.note:
print(f"说明：{tool.note}")
print("\n参数：")
for name, p in sig.parameters.items():
ann = "" if p.annotation is inspect.Parameter.empty else f": {p.annotation}"
default = "" if p.default is inspect.Parameter.empty else f" = {p.default!r}"
print(f" {name}{ann}{default}")
print(f"\n示例：python cli.py run {tool.name} " +
" ".join(f"{n}=..." for n, p in sig.parameters.items()
if p.default is inspect.Parameter.empty) + "\n")
return 0
def cmd_run(args) -> int:
tool = TOOLS_BY_NAME.get(args.tool)
if tool is None:
print(f"未找到工具：{args.tool}。用 `python cli.py list` 查看全部。", file=sys.stderr)
return 1
try:
func = _load_callable(tool)
except Exception as e:
print(f"工具 {tool.name} 所在模块导入失败（可能缺少可选依赖）：{e}", file=sys.stderr)
return 1
try:
kwargs = _parse_kwargs(func, args.params)
except Exception as e:
print(f"参数错误：{e}", file=sys.stderr)
return 1
print(f"调用工具 {tool.name} ...", file=sys.stderr)
try:
data = asyncio.run(_invoke(tool, kwargs))
except Exception as e:
print(f"调用失败：{type(e).__name__}: {e}", file=sys.stderr)
return 1
print(json.dumps(data, ensure_ascii=False, indent=2))
return 0 if data.get("success", True) else 2
# ---------------------------------------------------------------------------
# 端到端演示：一个「本地笔记 + 外部资料」研究助手 Agent 的感知流程
# ---------------------------------------------------------------------------
def _header(title: str) -> None:
print("\n" + "─" * 72)
print(f"▶ {title}")
print("─" * 72)
async def _demo(offline: bool) -> None:
from search_tools import search_web, search_knowledge_base
from filesystem_tools import grep_search, read_file
from public_data_tools import convert_currency, search_wikipedia
from multimodal_tools import read_webpage
# 各工具内部会用 logging.error 打印完整堆栈；演示时抬高阈值，让每步只显示
# CLI 自己组织的干净状态行（真实错误仍以友好提示呈现）。
logging.getLogger().setLevel(logging.CRITICAL)
print("\n" + "=" * 72)
print("感知工具端到端演示")
print("场景：一个研究助手 Agent 需要「先看本地资料、再补充外部信息」")
print(" 本演示串联五类感知工具，展示 Agent 如何『感知世界』" +
("（离线模式：跳过联网步骤）" if offline else ""))
print("=" * 72)
# 准备一个临时本地知识库，避免污染仓库
tmp = Path(tempfile.mkdtemp(prefix="perception_demo_"))
(tmp / "mcp_notes.md").write_text(
"# MCP 调研笔记\n\n"
"Model Context Protocol (MCP) 是一套开放协议，用于在 Agent 与工具/数据源之间\n"
"标准化上下文交换。感知工具（如 web_search、read_file）是 Agent 获取信息的感官。\n"
"关键设计：粒度权衡、输出信息量控制、上下文感知压缩。\n",
encoding="utf-8",
)
(tmp / "budget.md").write_text(
"# 预算\n\n本次调研的云资源预算为 200 USD，需要换算成人民币报销。\n",
encoding="utf-8",
)
# 1) 文件系统感知：在本地代码库里定位实现
_header("[1/5] 文件系统感知：grep 定位 + read_file 精读（离线可用）")
data = _unwrap(await grep_search("ActionResponse", str(SRC_DIR),
file_pattern="*.py", max_results=5))
if data.get("success"):
msg = data["message"]
print(f" grep 'ActionResponse' 命中 {msg['total_found']} 处，示例：")
for hit in msg["results"][:3]:
print(f" - {hit['file']}:{hit['line_number']}")
base_py = _unwrap(await read_file(str(SRC_DIR / "base.py"), max_length=200))
if base_py.get("success"):
head = base_py["message"]["content"].strip().splitlines()[0]
print(f" read_file base.py 首行：{head}")
# 2) 搜索感知：知识库检索（离线）+ 网络搜索（联网）
_header("[2/5] 搜索感知：本地知识库检索（离线）+ 网络搜索（联网）")
kb = _unwrap(await search_knowledge_base("MCP", str(tmp), top_k=3))
if kb.get("success"):
print(f" 知识库检索 'MCP' 命中 {kb['message']['total_found']} 个文件：")
for r in kb["message"]["results"]:
print(f" - {r['file']}（相关度 {r['relevance']}）")
if offline:
print(" 网络搜索：已跳过（离线模式）")
else:
try:
web = _unwrap(await search_web("Model Context Protocol", num_results=3))
if web.get("success") and web["message"]["results"]:
print(f" web_search 返回 {web['message']['count']} 条结果，首条：")
top = web["message"]["results"][0]
print(f" - {top['title']}\n {top['url']}")
else:
print(" web_search 未返回结果（可能被限流）")
except Exception as e:
print(f" web_search 失败（需要网络）：{e}")
# 3) 公开数据源感知：汇率换算（把预算 200 USD 换成 CNY）
_header("[3/5] 公开数据源感知：汇率换算 + Wikipedia 摘要（联网）")
if offline:
print(" 已跳过（离线模式）")
else:
try:
fx = _unwrap(await convert_currency(200, "USD", "CNY"))
if fx.get("success"):
m = fx["message"]
print(f" 预算换算：200 USD ≈ {m['converted_amount']:.2f} CNY"
f"（汇率 {m.get('exchange_rate')}）")
except Exception as e:
print(f" 汇率换算失败（需要网络）：{e}")
try:
wiki = _unwrap(await search_wikipedia("Model Context Protocol", sentences=2))
if wiki.get("success"):
print(f" Wikipedia：{wiki['message']['title']}")
print(f" {wiki['message']['summary'][:120]}...")
else:
print(" Wikipedia 未返回结果（可能被限流），Agent 可改用其它来源")
except Exception as e:
print(f" Wikipedia 查询失败（需要网络）：{e}")
# 4) 多模态理解：读取网页正文
_header("[4/5] 多模态理解：抓取网页正文（联网）")
if offline:
print(" 已跳过（离线模式）")
else:
try:
page = _unwrap(await read_webpage("https://example.com", extract_text=True))
if page.get("success"):
m = page["message"]
print(f" 网页标题：{m.get('title')}；正文长度：{m.get('text_length', 0)} 字符")
except Exception as e:
print(f" 网页抓取失败（需要网络）：{e}")
# 5) 私有数据源：需要授权
_header("[5/5] 私有数据源感知：日历 / Notion（需授权）")
print(" calendar_events 需 Google OAuth 授权，notion_search 需 NOTION_API_KEY。")
print(" 未配置时工具会返回结构化的失败信息，Agent 可据此提示用户去授权。")
print("\n" + "=" * 72)
print("演示完成。要点：感知工具是 Agent 的『感官』——只读、可缓存、可并行；")
print(" 设计关键在于粒度权衡与输出信息量控制（详见第四章）。")
print("=" * 72 + "\n")
# 清理临时知识库
for f in tmp.glob("*"):
f.unlink()
tmp.rmdir()
def cmd_demo(args) -> int:
asyncio.run(_demo(offline=args.offline))
return 0
# ---------------------------------------------------------------------------
# 参数解析
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="cli.py",
formatter_class=argparse.RawDescriptionHelpFormatter,
description="感知工具 MCP 服务器的命令行入口（实验 4-1）。\n"
"按五类感知场景组织：搜索 / 多模态理解 / 文件系统 / 公开数据源 / 私有数据源。",
epilog="示例：\n"
" python cli.py list 列出全部感知工具\n"
" python cli.py list --category filesystem 只看文件系统类\n"
" python cli.py info weather 查看 weather 的参数\n"
" python cli.py run grep pattern=async directory=src 调用 grep\n"
" python cli.py run currency_converter amount=100 from_currency=USD to_currency=CNY\n"
" python cli.py demo --offline 运行离线端到端演示\n",
)
sub = parser.add_subparsers(dest="command", required=True, metavar="<命令>")
p_list = sub.add_parser("list", help="列出全部感知工具（按五类分组）")
p_list.add_argument("--category", choices=list(CATEGORIES),
help="只列出某一类：" + " / ".join(f"{k}={v}" for k, v in CATEGORIES.items()))
p_list.set_defaults(handler=cmd_list)
p_info = sub.add_parser("info", help="查看某个工具的参数签名与示例")
p_info.add_argument("tool", help="工具名（见 list）")
p_info.set_defaults(handler=cmd_info)
p_run = sub.add_parser("run", help="直接调用某个工具并打印 JSON 结果")
p_run.add_argument("tool", help="工具名（见 list）")
p_run.add_argument("params", nargs="*", metavar="key=value",
help="以 key=value 形式传入的工具参数")
p_run.set_defaults(handler=cmd_run)
p_demo = sub.add_parser("demo", help="运行端到端感知场景演示")
p_demo.add_argument("--offline", action="store_true",
help="离线模式：只跑不联网的步骤（文件系统 / 本地知识库）")
p_demo.set_defaults(handler=cmd_demo)
return parser
def main(argv: list[str] | None = None) -> int:
logging.basicConfig(level=logging.WARNING,
format="%(levelname)s: %(message)s")
parser = build_parser()
args = parser.parse_args(argv)
return args.handler(args)
if __name__ == "__main__":
raise SystemExit(main())
quickstart.py¶
"""
Quick start script to test the perception tools MCP server.
"""
import asyncio
import json
import logging
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from search_tools import search_web, download_file
from multimodal_tools import read_webpage
from filesystem_tools import read_file, grep_search
from public_data_tools import get_weather, search_wikipedia, convert_currency
logging.basicConfig(level=logging.INFO)
async def test_tools():
"""Test various perception tools."""
print("\n" + "="*80)
print("PERCEPTION TOOLS MCP SERVER - QUICKSTART")
print("="*80)
# Test 1: Web Search
print("\n📝 Test 1: Web Search")
print("-" * 80)
try:
result = await search_web("Python programming", num_results=3)
data = json.loads(result.text)
if data['success']:
print(f"✅ Found {data['message']['count']} results")
if data['message']['results']:
for idx, result_item in enumerate(data['message']['results'], 1):
print(f"\n[{idx}] {result_item['title']}")
print(f" URL: {result_item['url']}")
if result_item.get('snippet'):
print(f" Snippet: {result_item['snippet']}")
else:
print(f"⚠️ Search API not configured: {data['message']}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 2: Wikipedia Search
print("\n📝 Test 2: Wikipedia Search")
print("-" * 80)
try:
result = await search_wikipedia("Artificial Intelligence", sentences=3)
data = json.loads(result.text)
if data['success']:
print(f"✅ Article: {data['message']['title']}")
print(f"Summary: {data['message']['summary'][:200]}...")
except Exception as e:
print(f"❌ Error: {e}")
# Test 3: Currency Conversion
print("\n📝 Test 3: Currency Conversion")
print("-" * 80)
try:
result = await convert_currency(100, "USD", "EUR")
data = json.loads(result.text)
if data['success']:
converted = data['message']['converted_amount']
print(f"✅ 100 USD = {converted:.2f} EUR")
except Exception as e:
print(f"❌ Error: {e}")
# Test 4: Weather
print("\n📝 Test 4: Weather Information")
print("-" * 80)
try:
result = await get_weather("London")
data = json.loads(result.text)
if data['success']:
temp = data['message']['temperature']
desc = data['message']['description']
print(f"✅ London: {temp}°C - {desc}")
else:
print(f"⚠️ Weather API not configured: {data['message']}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 5: Web Page Reading
print("\n📝 Test 5: Web Page Reading")
print("-" * 80)
try:
result = await read_webpage("https://www.example.com", extract_text=True)
data = json.loads(result.text)
if data['success']:
title = data['message']['title']
text_len = data['message'].get('text_length', 0)
print(f"✅ Page: {title}")
print(f"Text length: {text_len} characters")
except Exception as e:
print(f"❌ Error: {e}")
# Test 6: File Operations
print("\n📝 Test 6: File Operations (Reading this script)")
print("-" * 80)
try:
script_path = str(Path(__file__).resolve())
result = await read_file(script_path, max_length=500)
data = json.loads(result.text)
if data['success']:
size = data['message']['size_bytes']
print(f"✅ Read {size} bytes from {Path(script_path).name}")
except Exception as e:
print(f"❌ Error: {e}")
print("\n" + "="*80)
print("QUICKSTART COMPLETE")
print("="*80)
print("\nℹ️ Note: Some tests may fail if API keys are not configured.")
print(" Check env.example and configure your .env file for full functionality.")
print("\n")
if __name__ == "__main__":
asyncio.run(test_tools())
src/__init__.py¶
"""
Perception Tools MCP Server
A comprehensive MCP server for perception and data retrieval capabilities.
"""
__version__ = "1.0.0"
src/arxiv_enhanced.py¶
"""
Enhanced ArXiv tools with download and details.
Based on AWorld parxiv-server complete implementation.
"""
import json
import logging
import traceback
from typing import Union
import arxiv
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse
load_dotenv()
async def get_paper_details(
paper_id: str
) -> Union[str, TextContent]:
"""
Get detailed information about an ArXiv paper.
Args:
paper_id: ArXiv paper ID (e.g., '2301.07041')
Returns:
TextContent with paper details
"""
try:
clean_id = paper_id.replace("arxiv:", "").strip()
logging.info(f"📄 Getting paper details: {clean_id}")
search = arxiv.Search(id_list=[clean_id])
paper = next(arxiv.Client().results(search), None)
if not paper:
raise ValueError(f"Paper not found: {clean_id}")
result = {
"entry_id": paper.entry_id,
"title": paper.title,
"authors": [author.name for author in paper.authors],
"summary": paper.summary,
"published": paper.published.isoformat(),
"updated": paper.updated.isoformat() if paper.updated else None,
"categories": paper.categories,
"primary_category": paper.primary_category,
"pdf_url": paper.pdf_url,
"doi": paper.doi,
"journal_ref": paper.journal_ref
}
logging.info(f"✅ Retrieved paper: {paper.title}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"paper_id": clean_id}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to get paper details: {str(e)}"
logging.error(f"ArXiv error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "arxiv_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def download_paper(
paper_id: str,
download_dir: str = "."
) -> Union[str, TextContent]:
"""
Download ArXiv paper PDF.
Args:
paper_id: ArXiv paper ID
download_dir: Directory to save PDF
Returns:
TextContent with download result
"""
try:
from pathlib import Path
clean_id = paper_id.replace("arxiv:", "").strip()
logging.info(f"📥 Downloading paper: {clean_id}")
search = arxiv.Search(id_list=[clean_id])
paper = next(arxiv.Client().results(search), None)
if not paper:
raise ValueError(f"Paper not found: {clean_id}")
# Download
download_path = Path(download_dir)
download_path.mkdir(parents=True, exist_ok=True)
filename = f"{clean_id.replace('/', '_')}.pdf"
paper.download_pdf(dirpath=str(download_path), filename=filename)
file_path = download_path / filename
file_size = file_path.stat().st_size if file_path.exists() else 0
result = {
"paper_id": clean_id,
"title": paper.title,
"file_path": str(file_path),
"file_size": file_size,
"pdf_url": paper.pdf_url
}
logging.info(f"✅ Downloaded: {file_size} bytes")
action_response = ActionResponse(
success=True,
message=result,
metadata={"paper_id": clean_id}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Download failed: {str(e)}",
metadata={"error_type": "download_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_arxiv_categories() -> Union[str, TextContent]:
"""
Get list of ArXiv subject categories.
Returns:
TextContent with categories
"""
categories = {
"cs": "Computer Science",
"math": "Mathematics",
"physics": "Physics",
"astro-ph": "Astrophysics",
"cond-mat": "Condensed Matter",
"q-bio": "Quantitative Biology",
"q-fin": "Quantitative Finance",
"stat": "Statistics",
"econ": "Economics",
"eess": "Electrical Engineering"
}
result = {
"categories": categories,
"count": len(categories)
}
action_response = ActionResponse(
success=True,
message=result,
metadata={"total_categories": len(categories)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/base.py¶
"""
Base models and utilities for perception tools MCP server.
"""
import logging
import os
import tempfile
import traceback
from pathlib import Path
from urllib.parse import urlparse
from typing import Any
import requests
from pydantic import BaseModel, Field
class ActionResponse(BaseModel):
"""Standard response format for all perception tool actions."""
success: bool = Field(default=False, description="Whether the action was successfully executed")
message: Any = Field(default=None, description="The execution result of the action")
metadata: dict[str, Any] = Field(default_factory=dict, description="Additional metadata about the action")
class DocumentMetadata(BaseModel):
"""Metadata for document processing operations."""
file_name: str = Field(description="Original file name")
file_size: int = Field(description="File size in bytes")
file_type: str = Field(description="Document file type/extension")
absolute_path: str = Field(description="Absolute path to the document file")
page_count: int | None = Field(default=None, description="Number of pages in document")
processing_time: float | None = Field(default=None, description="Time taken to process")
output_format: str = Field(description="Format of the extracted content")
def is_url(path_or_url: str) -> bool:
"""
Check if the given string is a URL.
Args:
path_or_url: String to check
Returns:
True if the string is a URL, False otherwise
"""
parsed = urlparse(path_or_url)
return bool(parsed.scheme and parsed.netloc)
def validate_file_path(file_path: str) -> Path:
"""
Validate and resolve file path.
Args:
file_path: Path to the file
Returns:
Resolved Path object
Raises:
FileNotFoundError: If file doesn't exist
"""
path = Path(file_path).expanduser().resolve()
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
if not path.is_file():
raise ValueError(f"Path is not a file: {path}")
return path
def download_file_from_url(
url: str,
timeout: int = 60,
max_size_mb: float = 100.0
) -> tuple[str, bytes]:
"""
Download file from URL to temporary location.
Args:
url: URL to download from
timeout: Request timeout in seconds
max_size_mb: Maximum file size in MB
Returns:
Tuple of (temp_file_path, content)
Raises:
ValueError: If file size exceeds limit
requests.RequestException: If download fails
"""
max_size_bytes = max_size_mb * 1024 * 1024
try:
# Check content length first
head_response = requests.head(url, timeout=timeout, allow_redirects=True)
head_response.raise_for_status()
content_length = head_response.headers.get("content-length")
if content_length and int(content_length) > max_size_bytes:
raise ValueError(
f"File size ({int(content_length) / (1024 * 1024):.2f} MB) "
f"exceeds maximum allowed size ({max_size_mb} MB)"
)
# Download the file
response = requests.get(url, timeout=timeout, stream=True)
response.raise_for_status()
# Read content with size checking
content = b""
for chunk in response.iter_content(chunk_size=8192):
if len(content) + len(chunk) > max_size_bytes:
raise ValueError(f"File size exceeds maximum allowed size ({max_size_mb} MB)")
content += chunk
# Create temporary file
parsed_url = urlparse(url)
filename = os.path.basename(parsed_url.path) or "downloaded_file"
suffix = Path(filename).suffix or ".tmp"
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:
temp_file.write(content)
temp_path = temp_file.name
return temp_path, content
except requests.RequestException as e:
raise requests.RequestException(f"Failed to download file from URL: {e}")
except Exception as e:
raise IOError(f"Error downloading file: {e}")
src/document_processing_tools.py¶
"""
Document processing tools for PDF, DOCX, PPTX, CSV, TXT.
Based on AWorld MCP server implementation.
"""
import json
import logging
import traceback
from pathlib import Path
from typing import Union, Dict, Any
import pandas as pd
from docx import Document
from pptx import Presentation
import PyPDF2
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse, validate_file_path
load_dotenv()
async def extract_pdf_text(
file_path: str,
page_range: str | None = None
) -> Union[str, TextContent]:
"""
Extract text from PDF file.
Args:
file_path: Path to PDF file
page_range: Optional page range (e.g., "1-5" or "1,3,5")
Returns:
TextContent with extracted text
"""
try:
path = validate_file_path(file_path)
logging.info(f"📄 Extracting PDF: {path}")
with open(path, 'rb') as file:
reader = PyPDF2.PdfReader(file)
total_pages = len(reader.pages)
# Parse page range
if page_range:
pages_to_extract = parse_page_range(page_range, total_pages)
else:
pages_to_extract = range(total_pages)
# Extract text
text_parts = []
for page_num in pages_to_extract:
if page_num < total_pages:
page = reader.pages[page_num]
text = page.extract_text()
text_parts.append(f"--- Page {page_num + 1} ---\n{text}\n")
full_text = "\n".join(text_parts)
result = {
"file_name": path.name,
"file_type": "pdf",
"total_pages": total_pages,
"pages_extracted": len(pages_to_extract),
"text": full_text[:50000], # Limit to 50k chars
"text_length": len(full_text),
"truncated": len(full_text) > 50000
}
logging.info(f"✅ Extracted {len(pages_to_extract)} pages from PDF")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path), "pages": total_pages}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"PDF extraction failed: {str(e)}"
logging.error(f"PDF error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "pdf_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_docx_content(
file_path: str
) -> Union[str, TextContent]:
"""
Extract content from DOCX file.
Args:
file_path: Path to DOCX file
Returns:
TextContent with extracted content
"""
try:
path = validate_file_path(file_path)
logging.info(f"📄 Extracting DOCX: {path}")
doc = Document(path)
# Extract paragraphs
paragraphs = [para.text for para in doc.paragraphs if para.text.strip()]
# Extract tables
tables_data = []
for table in doc.tables:
table_data = []
for row in table.rows:
row_data = [cell.text for cell in row.cells]
table_data.append(row_data)
tables_data.append(table_data)
full_text = "\n\n".join(paragraphs)
result = {
"file_name": path.name,
"file_type": "docx",
"paragraphs": len(paragraphs),
"tables": len(tables_data),
"text": full_text[:50000],
"text_length": len(full_text),
"truncated": len(full_text) > 50000,
"tables_data": tables_data if tables_data else []
}
logging.info(f"✅ Extracted DOCX: {len(paragraphs)} paragraphs, {len(tables_data)} tables")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"DOCX extraction failed: {str(e)}"
logging.error(f"DOCX error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "docx_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_pptx_content(
file_path: str
) -> Union[str, TextContent]:
"""
Extract content from PPTX file.
Args:
file_path: Path to PPTX file
Returns:
TextContent with extracted content
"""
try:
path = validate_file_path(file_path)
logging.info(f"📊 Extracting PPTX: {path}")
prs = Presentation(path)
slides_content = []
for slide_num, slide in enumerate(prs.slides, 1):
slide_text = []
for shape in slide.shapes:
if hasattr(shape, "text") and shape.text.strip():
slide_text.append(shape.text)
if slide_text:
slides_content.append({
"slide_number": slide_num,
"text": "\n".join(slide_text)
})
full_text = "\n\n".join([f"=== Slide {s['slide_number']} ===\n{s['text']}" for s in slides_content])
result = {
"file_name": path.name,
"file_type": "pptx",
"total_slides": len(prs.slides),
"slides_with_content": len(slides_content),
"text": full_text[:50000],
"text_length": len(full_text),
"truncated": len(full_text) > 50000,
"slides": slides_content
}
logging.info(f"✅ Extracted PPTX: {len(prs.slides)} slides")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"PPTX extraction failed: {str(e)}"
logging.error(f"PPTX error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "pptx_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_csv_content(
file_path: str,
max_rows: int = 1000
) -> Union[str, TextContent]:
"""
Extract and parse CSV file content.
Args:
file_path: Path to CSV file
max_rows: Maximum rows to read
Returns:
TextContent with parsed CSV data
"""
try:
path = validate_file_path(file_path)
logging.info(f"📊 Parsing CSV: {path}")
# Read CSV with pandas
df = pd.read_csv(path, nrows=max_rows)
result = {
"file_name": path.name,
"file_type": "csv",
"rows": len(df),
"columns": len(df.columns),
"column_names": df.columns.tolist(),
"data": df.head(100).to_dict(orient="records"), # First 100 rows
"preview": df.head(10).to_string(),
"truncated": len(df) == max_rows
}
logging.info(f"✅ Parsed CSV: {len(df)} rows, {len(df.columns)} columns")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"CSV parsing failed: {str(e)}"
logging.error(f"CSV error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "csv_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
def parse_page_range(page_range: str, total_pages: int) -> list[int]:
"""
Parse page range string into list of page numbers.
Args:
page_range: String like "1-5" or "1,3,5" or "1-3,7,9-11"
total_pages: Total number of pages
Returns:
List of page numbers (0-indexed)
"""
pages = []
for part in page_range.split(","):
part = part.strip()
if "-" in part:
start, end = map(int, part.split("-"))
pages.extend(range(start - 1, min(end, total_pages)))
else:
page_num = int(part) - 1
if 0 <= page_num < total_pages:
pages.append(page_num)
return sorted(set(pages))
src/filesystem_tools.py¶
"""
File system tools: file reading, grep search, and text summarization.
"""
import json
import logging
import os
import re
import subprocess
import traceback
from pathlib import Path
from typing import Union
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse, validate_file_path
load_dotenv()
async def read_file(
file_path: str,
encoding: str = "utf-8",
max_length: int = 50000
) -> Union[str, TextContent]:
"""
Read a file and return its contents.
Args:
file_path: Path to the file
encoding: File encoding (default: utf-8)
max_length: Maximum number of characters to return
Returns:
TextContent with file contents
"""
try:
path = validate_file_path(file_path)
logging.info(f"📖 Reading file: {path}")
with open(path, 'r', encoding=encoding, errors='ignore') as f:
content = f.read()
truncated = len(content) > max_length
if truncated:
content = content[:max_length]
result = {
"file_path": str(path),
"content": content,
"size_bytes": path.stat().st_size,
"truncated": truncated,
"encoding": encoding
}
logging.info(f"✅ Successfully read file ({len(content)} characters)")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"File reading failed: {str(e)}"
logging.error(f"File read error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "file_read_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def grep_search(
pattern: str,
directory: str,
file_pattern: str = "*",
recursive: bool = True,
case_sensitive: bool = False,
max_results: int = 100
) -> Union[str, TextContent]:
"""
Search for a pattern in files using grep-like functionality.
Args:
pattern: Regular expression pattern to search for
directory: Directory to search in
file_pattern: File pattern to match (e.g., "*.py")
recursive: Whether to search recursively
case_sensitive: Whether search is case-sensitive
max_results: Maximum number of results to return
Returns:
TextContent with search results
"""
try:
dir_path = Path(directory).expanduser().resolve()
if not dir_path.exists():
raise FileNotFoundError(f"Directory not found: {dir_path}")
if not dir_path.is_dir():
raise ValueError(f"Path is not a directory: {dir_path}")
logging.info(f"🔍 Searching for pattern '{pattern}' in {dir_path}")
results = []
flags = re.IGNORECASE if not case_sensitive else 0
regex = re.compile(pattern, flags)
# Determine which files to search
if recursive:
files = dir_path.rglob(file_pattern)
else:
files = dir_path.glob(file_pattern)
for file_path in files:
if not file_path.is_file():
continue
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
for line_num, line in enumerate(f, 1):
if regex.search(line):
results.append({
"file": str(file_path.relative_to(dir_path)),
"line_number": line_num,
"line": line.strip(),
"absolute_path": str(file_path)
})
if len(results) >= max_results:
break
if len(results) >= max_results:
break
except Exception as e:
logging.warning(f"Error reading {file_path}: {e}")
continue
logging.info(f"✅ Found {len(results)} matches")
action_response = ActionResponse(
success=True,
message={
"pattern": pattern,
"results": results,
"total_found": len(results),
"truncated": len(results) >= max_results
},
metadata={
"directory": str(dir_path),
"file_pattern": file_pattern,
"recursive": recursive
}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Grep search failed: {str(e)}"
logging.error(f"Grep error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "grep_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def summarize_text(
text: str,
max_length: int = 500,
use_llm: bool = True
) -> Union[str, TextContent]:
"""
Summarize long text content.
Args:
text: Text to summarize
max_length: Target summary length
use_llm: Whether to use LLM for summarization (if available)
Returns:
TextContent with summary
"""
try:
logging.info(f"📝 Summarizing text ({len(text)} characters)")
if use_llm:
# TODO: Integrate with LLM API for better summarization
# For now, use simple extraction
summary = "LLM summarization not yet implemented. Using simple extraction."
method = "placeholder"
else:
# Simple extractive summarization: first N sentences
sentences = re.split(r'[.!?]+', text)
summary = ""
for sentence in sentences:
if len(summary) + len(sentence) > max_length:
break
summary += sentence.strip() + ". "
method = "extractive"
if not summary or summary == "LLM summarization not yet implemented. Using simple extraction.":
# Fallback: just truncate
summary = text[:max_length] + "..." if len(text) > max_length else text
method = "truncation"
result = {
"original_length": len(text),
"summary_length": len(summary),
"summary": summary,
"method": method,
"compression_ratio": len(summary) / len(text) if len(text) > 0 else 0
}
logging.info(f"✅ Generated summary ({len(summary)} characters)")
action_response = ActionResponse(
success=True,
message=result,
metadata={"method": method}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Text summarization failed: {str(e)}"
logging.error(f"Summarization error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "summarization_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/google_search_enhanced.py¶
"""
Enhanced Google Search tools.
Based on AWorld google-search server.
"""
import json
import logging
import os
import traceback
from typing import Union
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse
load_dotenv()
async def google_search_api(
query: str,
num_results: int = 5,
safe_search: bool = True,
language: str = "en",
country: str = "us"
) -> Union[str, TextContent]:
"""
Search Google using Custom Search API.
Args:
query: Search query
num_results: Number of results (1-10)
safe_search: Enable safe search
language: Language code
country: Country code
Returns:
TextContent with search results
"""
try:
api_key = os.getenv("GOOGLE_API_KEY")
cse_id = os.getenv("GOOGLE_CSE_ID")
if not api_key or not cse_id:
return await _fallback_google_search(query, num_results)
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": api_key,
"cx": cse_id,
"q": query,
"num": min(num_results, 10),
"safe": "active" if safe_search else "off",
"hl": language,
"gl": country
}
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
results = []
if "items" in data:
for item in data["items"]:
results.append({
"title": item.get("title"),
"url": item.get("link"),
"snippet": item.get("snippet"),
"display_url": item.get("displayLink")
})
action_response = ActionResponse(
success=True,
message={"query": query, "results": results, "count": len(results)},
metadata={"engine": "google_api", "results_count": len(results)}
)
logging.info(f"✅ Google Search: {len(results)} results")
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
logging.error(f"Google API search failed: {traceback.format_exc()}")
return await _fallback_google_search(query, num_results)
async def _fallback_google_search(query: str, num_results: int) -> TextContent:
"""Fallback to DuckDuckGo if Google API not available."""
try:
logging.info("Using DuckDuckGo fallback")
url = "https://html.duckduckgo.com/html/"
headers = {"User-Agent": "Mozilla/5.0"}
data = {"q": query}
response = requests.post(url, data=data, headers=headers, timeout=15)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
result_divs = soup.find_all('div', class_='result')
results = []
for i, div in enumerate(result_divs[:num_results]):
title_tag = div.find('a', class_='result__a')
if title_tag:
results.append({
"title": title_tag.get_text(strip=True),
"url": title_tag.get('href', ''),
"snippet": div.find('a', class_='result__snippet').get_text(strip=True) if div.find('a', class_='result__snippet') else ""
})
action_response = ActionResponse(
success=True,
message={"query": query, "results": results, "count": len(results)},
metadata={"engine": "duckduckgo_fallback"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Search failed: {str(e)}",
metadata={"error_type": "search_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def read_webpage_content(
url: str,
extract_links: bool = False
) -> Union[str, TextContent]:
"""
Read and extract content from webpage.
Args:
url: URL to read
extract_links: Whether to extract links
Returns:
TextContent with webpage content
"""
try:
headers = {"User-Agent": "Mozilla/5.0"}
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# Remove scripts and styles
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
text = ' '.join(line for line in lines if line)
result = {
"url": url,
"title": soup.title.string if soup.title else "No title",
"text": text[:10000],
"text_length": len(text)
}
if extract_links:
links = []
for link in soup.find_all('a', href=True)[:50]:
links.append({
"text": link.get_text().strip(),
"href": link['href']
})
result["links"] = links
action_response = ActionResponse(
success=True,
message=result,
metadata={"url": url}
)
logging.info(f"✅ Read webpage: {len(text)} chars")
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Failed to read webpage: {str(e)}",
metadata={"error_type": "webpage_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/main.py¶
"""
Main MCP server for perception tools.
This MCP server provides comprehensive perception capabilities including:
- Search tools (web search, knowledge base, file download)
- Multimodal understanding (web pages, documents, images, videos)
- File system operations (read, grep, summarization)
- Public data sources (weather, stocks, currency, Wikipedia, ArXiv, Wayback)
- Private data sources (Google Calendar, Notion)
"""
import logging
from dotenv import load_dotenv
from mcp.server import FastMCP
from pydantic import Field
# Import all tool functions
from search_tools import search_web, download_file, search_knowledge_base
from multimodal_tools import read_webpage, read_document, parse_image, parse_video, extract_youtube_transcript, download_youtube_video
from filesystem_tools import read_file, grep_search, summarize_text
from public_data_tools import (
get_weather, get_stock_price, convert_currency,
search_wikipedia, search_arxiv, search_wayback,
get_crypto_price, search_location, search_poi
)
from private_data_tools import get_calendar_events, search_notion
from pubchem_tools import search_compounds, get_compound_properties, get_compound_synonyms, search_similar_compounds
from yahoo_finance_tools import get_stock_quote, get_historical_data, get_company_info, get_financial_statements
from document_processing_tools import extract_pdf_text, extract_docx_content, extract_pptx_content, extract_csv_content
from media_processing_tools import transcribe_audio_whisper, extract_audio_metadata, extract_text_ocr, analyze_image_ai, extract_video_keyframes, analyze_video_ai, trim_audio, get_image_metadata
from google_search_enhanced import google_search_api, read_webpage_content
from wiki_enhanced import get_article_content, get_article_categories, get_article_links, get_article_history
from arxiv_enhanced import get_paper_details, download_paper, get_arxiv_categories
from wayback_enhanced import get_archived_content
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
load_dotenv()
# Initialize MCP server
mcp = FastMCP(
"perception-tools",
instructions="""
Perception Tools MCP Server
A comprehensive MCP server providing various perception and data retrieval capabilities:
## Search Tools
- Web search using DuckDuckGo (free, no API key required)
- Local knowledge base search
- File download from URLs
## Multimodal Understanding
- Web page content extraction
- Document reading (PDF, DOCX, PPTX)
- Image parsing and analysis
- Video metadata extraction
## File System Tools
- File reading with encoding support
- Grep-like pattern search
- Text summarization
## Public Data Sources
- Weather information
- Stock prices and market data
- Cryptocurrency prices (CoinGecko)
- Currency conversion
- Location search / geocoding (Nominatim)
- Points of Interest search (Overpass)
- Wikipedia search
- ArXiv academic papers
- Wayback Machine archives
## Private Data Sources
- Google Calendar events
- Notion workspace search
"""
)
# ============================================================================
# SEARCH TOOLS
# ============================================================================
@mcp.tool(description="Search the web using DuckDuckGo (free, no API key required)")
async def web_search(
query: str = Field(description="Search query string"),
num_results: int = Field(default=5, description="Number of results (1-10)"),
region: str = Field(default="wt-wt", description="Region code (e.g., 'us-en', 'uk-en', 'wt-wt' for worldwide)")
):
"""Search the web and return results."""
return await search_web(query, num_results, region)
@mcp.tool(description="Download a file from a URL to local storage")
async def download(
url: str = Field(description="URL to download from"),
output_path: str = Field(description="Local path to save the file"),
overwrite: bool = Field(default=False, description="Overwrite existing file"),
timeout: int = Field(default=180, description="Download timeout in seconds")
):
"""Download a file from URL."""
return await download_file(url, output_path, overwrite, timeout)
@mcp.tool(description="Search a local knowledge base directory")
async def knowledge_base_search(
query: str = Field(description="Search query"),
knowledge_base_path: str = Field(description="Path to knowledge base directory"),
top_k: int = Field(default=5, description="Number of top results")
):
"""Search local knowledge base."""
return await search_knowledge_base(query, knowledge_base_path, top_k)
# ============================================================================
# MULTIMODAL UNDERSTANDING TOOLS
# ============================================================================
@mcp.tool(description="Read and extract content from a webpage")
async def webpage_reader(
url: str = Field(description="URL of the webpage"),
extract_text: bool = Field(default=True, description="Extract text content"),
extract_links: bool = Field(default=False, description="Extract links")
):
"""Read webpage content."""
return await read_webpage(url, extract_text, extract_links)
@mcp.tool(description="Read and extract content from documents (PDF, DOCX, PPTX)")
async def document_reader(
file_path: str = Field(description="Path to document file or URL"),
extract_images: bool = Field(default=False, description="Extract images")
):
"""Read document content."""
return await read_document(file_path, extract_images)
@mcp.tool(description="Parse and analyze image files")
async def image_parser(
image_path: str = Field(description="Path to image file or URL"),
use_llm: bool = Field(default=True, description="Use LLM for analysis")
):
"""Parse image content."""
return await parse_image(image_path, use_llm)
@mcp.tool(description="Parse and extract metadata from video files")
async def video_parser(
video_path: str = Field(description="Path to video file or URL"),
extract_frames: bool = Field(default=False, description="Extract sample frames"),
frame_interval: int = Field(default=30, description="Frame extraction interval")
):
"""Parse video metadata."""
return await parse_video(video_path, extract_frames, frame_interval)
# ============================================================================
# FILE SYSTEM TOOLS
# ============================================================================
@mcp.tool(description="Read a file and return its contents")
async def file_reader(
file_path: str = Field(description="Path to the file"),
encoding: str = Field(default="utf-8", description="File encoding"),
max_length: int = Field(default=50000, description="Maximum characters to read")
):
"""Read file contents."""
return await read_file(file_path, encoding, max_length)
@mcp.tool(description="Search for patterns in files (grep-like functionality)")
async def grep(
pattern: str = Field(description="Regular expression pattern"),
directory: str = Field(description="Directory to search in"),
file_pattern: str = Field(default="*", description="File pattern (e.g., *.py)"),
recursive: bool = Field(default=True, description="Search recursively"),
case_sensitive: bool = Field(default=False, description="Case-sensitive search"),
max_results: int = Field(default=100, description="Maximum results")
):
"""Search files for pattern."""
return await grep_search(pattern, directory, file_pattern, recursive, case_sensitive, max_results)
@mcp.tool(description="Summarize long text content")
async def text_summarizer(
text: str = Field(description="Text to summarize"),
max_length: int = Field(default=500, description="Target summary length"),
use_llm: bool = Field(default=True, description="Use LLM for summarization")
):
"""Summarize text."""
return await summarize_text(text, max_length, use_llm)
# ============================================================================
# PUBLIC DATA SOURCE TOOLS
# ============================================================================
@mcp.tool(description="Get current weather information for a location (Open-Meteo, free, no API key)")
async def weather(
location: str = Field(description="City name (automatically geocoded)"),
latitude: float | None = Field(default=None, description="Latitude coordinate (optional)"),
longitude: float | None = Field(default=None, description="Longitude coordinate (optional)")
):
"""Get weather data."""
return await get_weather(location, latitude, longitude)
@mcp.tool(description="Get stock price and market information")
async def stock_price(
symbol: str = Field(description="Stock ticker symbol (e.g., AAPL)"),
interval: str = Field(default="1d", description="Data interval")
):
"""Get stock price."""
return await get_stock_price(symbol, interval)
@mcp.tool(description="Convert between currencies")
async def currency_converter(
amount: float = Field(description="Amount to convert"),
from_currency: str = Field(description="Source currency code (e.g., USD)"),
to_currency: str = Field(description="Target currency code (e.g., EUR)")
):
"""Convert currency."""
return await convert_currency(amount, from_currency, to_currency)
@mcp.tool(description="Get cryptocurrency price information (CoinGecko, free, no API key)")
async def crypto_price(
symbol: str = Field(description="Cryptocurrency symbol or ID (e.g., bitcoin, ethereum, btc, eth)"),
vs_currency: str = Field(default="usd", description="Target currency (usd, eur, gbp, etc.)")
):
"""Get cryptocurrency price."""
return await get_crypto_price(symbol, vs_currency)
@mcp.tool(description="Search for locations using Nominatim/OpenStreetMap (free, no API key)")
async def location_search(
query: str = Field(description="Location query (e.g., 'Eiffel Tower', 'New York', 'Tokyo')"),
limit: int = Field(default=5, description="Maximum number of results (1-50)"),
country_code: str | None = Field(default=None, description="Country code filter (e.g., 'us', 'gb', 'fr')")
):
"""Search locations (geocoding)."""
return await search_location(query, limit, country_code)
@mcp.tool(description="Search for Points of Interest near a location using Overpass/OpenStreetMap (free, no API key)")
async def poi_search(
query: str = Field(description="Type of POI (e.g., 'restaurant', 'cafe', 'hospital', 'atm', 'hotel')"),
latitude: float = Field(description="Center latitude coordinate"),
longitude: float = Field(description="Center longitude coordinate"),
radius: int = Field(default=1000, description="Search radius in meters"),
limit: int = Field(default=10, description="Maximum number of results")
):
"""Search points of interest."""
return await search_poi(query, latitude, longitude, radius, limit)
@mcp.tool(description="Search Wikipedia and get article summary")
async def wikipedia_search(
query: str = Field(description="Search query"),
language: str = Field(default="en", description="Wikipedia language"),
sentences: int = Field(default=5, description="Summary sentence count")
):
"""Search Wikipedia."""
return await search_wikipedia(query, language, sentences)
@mcp.tool(description="Search ArXiv for academic papers")
async def arxiv_search(
query: str = Field(description="Search query"),
max_results: int = Field(default=5, description="Maximum results"),
sort_by: str = Field(default="relevance", description="Sort method")
):
"""Search ArXiv."""
return await search_arxiv(query, max_results, sort_by)
@mcp.tool(description="Search Wayback Machine for archived web pages")
async def wayback_search(
url: str = Field(description="URL to search for"),
year: int | None = Field(default=None, description="Filter by year"),
limit: int = Field(default=10, description="Maximum snapshots")
):
"""Search Wayback Machine."""
return await search_wayback(url, year, limit)
# ============================================================================
# YOUTUBE TOOLS
# ============================================================================
@mcp.tool(description="Extract transcript from a YouTube video")
async def youtube_transcript(
video_id: str = Field(description="YouTube video ID or URL"),
language_code: str = Field(default="en", description="Language code for transcript"),
translate_to_language: str | None = Field(default=None, description="Translate to this language")
):
"""Extract YouTube transcript."""
return await extract_youtube_transcript(video_id, language_code, translate_to_language)
# ============================================================================
# PUBCHEM CHEMICAL DATA TOOLS
# ============================================================================
@mcp.tool(description="Search PubChem for chemical compounds")
async def pubchem_search(
query: str = Field(description="Search term or identifier"),
search_type: str = Field(default="name", description="Type: name, cid, smiles, inchi, formula"),
max_results: int = Field(default=10, description="Maximum results (1-100)")
):
"""Search PubChem compounds."""
return await search_compounds(query, search_type, max_results)
@mcp.tool(description="Get detailed properties for a PubChem compound")
async def pubchem_properties(
cid: int = Field(description="PubChem Compound ID"),
properties: list[str] | None = Field(default=None, description="List of property names")
):
"""Get compound properties."""
return await get_compound_properties(cid, properties)
@mcp.tool(description="Get synonyms for a PubChem compound")
async def pubchem_synonyms(
cid: int = Field(description="PubChem Compound ID"),
max_synonyms: int = Field(default=20, description="Maximum synonyms (1-100)")
):
"""Get compound synonyms."""
return await get_compound_synonyms(cid, max_synonyms)
@mcp.tool(description="Search for structurally similar compounds in PubChem")
async def pubchem_similar(
cid: int = Field(description="Reference compound CID"),
similarity_threshold: float = Field(default=0.9, description="Similarity threshold (0.0-1.0)"),
max_results: int = Field(default=10, description="Maximum results (1-50)")
):
"""Search similar compounds."""
return await search_similar_compounds(cid, similarity_threshold, max_results)
# ============================================================================
# YAHOO FINANCE TOOLS
# ============================================================================
@mcp.tool(description="Get current stock quote and market data")
async def yfinance_quote(
symbol: str = Field(description="Stock ticker symbol (e.g., AAPL, MSFT)")
):
"""Get stock quote."""
return await get_stock_quote(symbol)
@mcp.tool(description="Get historical stock price data")
async def yfinance_historical(
symbol: str = Field(description="Stock ticker symbol"),
start: str = Field(description="Start date (YYYY-MM-DD)"),
end: str = Field(description="End date (YYYY-MM-DD)"),
interval: str = Field(default="1d", description="Data interval (1d, 1wk, 1mo)"),
max_rows_preview: int = Field(default=10, description="Max rows in preview")
):
"""Get historical stock data."""
return await get_historical_data(symbol, start, end, interval, max_rows_preview)
@mcp.tool(description="Get comprehensive company information")
async def yfinance_company_info(
symbol: str = Field(description="Stock ticker symbol")
):
"""Get company information."""
return await get_company_info(symbol)
@mcp.tool(description="Get financial statements (income statement, balance sheet, cash flow)")
async def yfinance_financials(
symbol: str = Field(description="Stock ticker symbol"),
statement_type: str = Field(description="Type: income_statement, balance_sheet, cash_flow"),
period_type: str = Field(default="annual", description="Period: annual or quarterly"),
max_columns_preview: int = Field(default=4, description="Max periods to show")
):
"""Get financial statements."""
return await get_financial_statements(symbol, statement_type, period_type, max_columns_preview)
# ============================================================================
# DOCUMENT PROCESSING TOOLS
# ============================================================================
@mcp.tool(description="Extract text from PDF file with optional page range")
async def pdf_extract(
file_path: str = Field(description="Path to PDF file"),
page_range: str | None = Field(default=None, description="Page range (e.g., '1-5' or '1,3,5')")
):
"""Extract text from PDF."""
return await extract_pdf_text(file_path, page_range)
@mcp.tool(description="Extract content from Word document (DOCX)")
async def docx_extract(
file_path: str = Field(description="Path to DOCX file")
):
"""Extract content from DOCX."""
return await extract_docx_content(file_path)
@mcp.tool(description="Extract content from PowerPoint presentation (PPTX)")
async def pptx_extract(
file_path: str = Field(description="Path to PPTX file")
):
"""Extract content from PPTX."""
return await extract_pptx_content(file_path)
@mcp.tool(description="Extract and parse CSV file data")
async def csv_parse(
file_path: str = Field(description="Path to CSV file"),
max_rows: int = Field(default=1000, description="Maximum rows to read")
):
"""Parse CSV data."""
return await extract_csv_content(file_path, max_rows)
# ============================================================================
# MEDIA PROCESSING TOOLS
# ============================================================================
@mcp.tool(description="Transcribe audio to text using Whisper")
async def audio_transcribe(
file_path: str = Field(description="Path to audio file"),
model_size: str = Field(default="base", description="Whisper model size"),
language: str = Field(default="en", description="Language code")
):
"""Transcribe audio to text."""
return await transcribe_audio_whisper(file_path, model_size, language)
@mcp.tool(description="Extract audio file metadata")
async def audio_metadata(
file_path: str = Field(description="Path to audio file")
):
"""Extract audio metadata."""
return await extract_audio_metadata(file_path)
@mcp.tool(description="Extract text from image using OCR")
async def image_ocr(
image_path: str = Field(description="Path to image file"),
language: str = Field(default="eng", description="OCR language")
):
"""Extract text from image using OCR."""
return await extract_text_ocr(image_path, language)
@mcp.tool(description="Analyze image using AI vision")
async def image_analyze(
image_path: str = Field(description="Path to image file"),
prompt: str = Field(default="Describe this image in detail", description="Analysis prompt")
):
"""Analyze image with AI."""
return await analyze_image_ai(image_path, prompt)
@mcp.tool(description="Extract keyframes from video")
async def video_keyframes(
video_path: str = Field(description="Path to video file"),
num_frames: int = Field(default=10, description="Number of keyframes to extract")
):
"""Extract video keyframes."""
return await extract_video_keyframes(video_path, num_frames)
@mcp.tool(description="Analyze video content using AI vision")
async def video_analyze(
video_path: str = Field(description="Path to video file"),
num_frames: int = Field(default=5, description="Number of frames to analyze"),
prompt: str = Field(default="Analyze this video and describe what's happening", description="Analysis prompt")
):
"""Analyze video with AI."""
return await analyze_video_ai(video_path, num_frames, prompt)
@mcp.tool(description="Trim audio file to specific time range")
async def audio_trim(
audio_path: str = Field(description="Path to audio file"),
start_time: float = Field(description="Start time in seconds"),
duration: float | None = Field(default=None, description="Duration in seconds"),
output_path: str | None = Field(default=None, description="Output file path")
):
"""Trim audio file."""
return await trim_audio(audio_path, start_time, duration, output_path)
@mcp.tool(description="Get detailed image metadata including EXIF")
async def image_metadata(
image_path: str = Field(description="Path to image file")
):
"""Get image metadata."""
return await get_image_metadata(image_path)
@mcp.tool(description="Download YouTube video")
async def youtube_download(
url: str = Field(description="YouTube video URL"),
output_dir: str = Field(default=".", description="Output directory"),
max_resolution: str = Field(default="720p", description="Maximum resolution")
):
"""Download YouTube video."""
return await download_youtube_video(url, output_dir, max_resolution)
# ============================================================================
# GOOGLE SEARCH ENHANCED TOOLS
# ============================================================================
@mcp.tool(description="Search Google with API or DuckDuckGo fallback")
async def google_search_enhanced(
query: str = Field(description="Search query"),
num_results: int = Field(default=5, description="Number of results (1-10)"),
safe_search: bool = Field(default=True, description="Enable safe search"),
language: str = Field(default="en", description="Language code"),
country: str = Field(default="us", description="Country code")
):
"""Enhanced Google search."""
return await google_search_api(query, num_results, safe_search, language, country)
@mcp.tool(description="Read and extract content from webpage")
async def webpage_read_enhanced(
url: str = Field(description="URL to read"),
extract_links: bool = Field(default=False, description="Extract links from page")
):
"""Read webpage content."""
return await read_webpage_content(url, extract_links)
# ============================================================================
# WIKIPEDIA ENHANCED TOOLS
# ============================================================================
@mcp.tool(description="Get full Wikipedia article content")
async def wiki_article_full(
title: str = Field(description="Article title"),
language: str = Field(default="en", description="Language code")
):
"""Get full Wikipedia article."""
return await get_article_content(title, language)
@mcp.tool(description="Get Wikipedia article categories")
async def wiki_article_categories(
title: str = Field(description="Article title"),
language: str = Field(default="en", description="Language code")
):
"""Get article categories."""
return await get_article_categories(title, language)
@mcp.tool(description="Get links from Wikipedia article")
async def wiki_article_links(
title: str = Field(description="Article title"),
language: str = Field(default="en", description="Language code")
):
"""Get article links."""
return await get_article_links(title, language)
@mcp.tool(description="Get historical version of Wikipedia article")
async def wiki_article_history(
title: str = Field(description="Article title"),
date: str = Field(description="Date (YYYY/MM/DD)"),
language: str = Field(default="en", description="Language code")
):
"""Get historical Wikipedia article."""
return await get_article_history(title, date, language)
# ============================================================================
# ARXIV ENHANCED TOOLS
# ============================================================================
@mcp.tool(description="Get detailed ArXiv paper information")
async def arxiv_paper_details(
paper_id: str = Field(description="ArXiv paper ID")
):
"""Get paper details."""
return await get_paper_details(paper_id)
@mcp.tool(description="Download ArXiv paper PDF")
async def arxiv_download(
paper_id: str = Field(description="ArXiv paper ID"),
download_dir: str = Field(default=".", description="Download directory")
):
"""Download ArXiv paper."""
return await download_paper(paper_id, download_dir)
@mcp.tool(description="Get ArXiv subject categories")
async def arxiv_categories():
"""Get ArXiv categories."""
return await get_arxiv_categories()
# ============================================================================
# WAYBACK ENHANCED TOOLS
# ============================================================================
@mcp.tool(description="Get content from archived webpage")
async def wayback_archived_content(
url: str = Field(description="URL to retrieve"),
timestamp: str = Field(description="Timestamp (YYYYMMDDhhmmss)")
):
"""Get archived webpage content."""
return await get_archived_content(url, timestamp)
# ============================================================================
# PRIVATE DATA SOURCE TOOLS
# ============================================================================
@mcp.tool(description="Get events from Google Calendar")
async def calendar_events(
start_date: str | None = Field(default=None, description="Start date (ISO format)"),
end_date: str | None = Field(default=None, description="End date (ISO format)"),
calendar_id: str = Field(default="primary", description="Calendar ID"),
max_results: int = Field(default=10, description="Maximum events")
):
"""Get calendar events."""
return await get_calendar_events(start_date, end_date, calendar_id, max_results)
@mcp.tool(description="Search Notion workspace")
async def notion_search(
query: str = Field(description="Search query"),
database_id: str | None = Field(default=None, description="Specific database ID"),
page_size: int = Field(default=10, description="Results per page")
):
"""Search Notion."""
return await search_notion(query, database_id, page_size)
# ============================================================================
# RUN SERVER
# ============================================================================
if __name__ == "__main__":
logging.info("Starting Perception Tools MCP server!")
mcp.run(transport="stdio")
src/media_processing_tools.py¶
"""
Media processing tools for audio, image, and video.
Based on AWorld MCP server implementation.
"""
import json
import logging
import traceback
import subprocess
import base64
from pathlib import Path
from typing import Union, Dict, Any
import cv2
from PIL import Image
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse, validate_file_path
load_dotenv()
def _map_model_for_openrouter(model: str) -> str:
"""Map a plain model id onto OpenRouter's `provider/model` form."""
if "/" in model:
return model
m = model.lower()
if m.startswith(("gpt-", "o1-", "o3-", "o4-")):
return f"openai/{model}"
if m.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return model
def _make_vision_client(default_model: str = "gpt-5.6-luna"):
"""Build an OpenAI-compatible vision client with a universal fallback.
Preferred path uses OPENAI_API_KEY directly. When it is absent but an
OPENROUTER_API_KEY is set, transparently route through OpenRouter (mapping
the model id to provider/model form) so the vision tools still run.
Returns (client, model). Raises ValueError with the accepted keys listed
when neither credential is available.
"""
import os
from openai import OpenAI
model = os.getenv("PERCEPTION_VISION_MODEL", default_model)
or_key = os.getenv("OPENROUTER_API_KEY")
# gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
# when an OpenRouter key is present, prefer routing these ids through it.
if or_key and model.lower().startswith("gpt-5"):
client = OpenAI(api_key=or_key, base_url="https://openrouter.ai/api/v1")
return client, _map_model_for_openrouter(model)
api_key = os.getenv("OPENAI_API_KEY")
if api_key:
base_url = os.getenv("OPENAI_BASE_URL")
client = OpenAI(api_key=api_key, base_url=base_url) if base_url else OpenAI(api_key=api_key)
return client, model
if or_key:
client = OpenAI(api_key=or_key, base_url="https://openrouter.ai/api/v1")
return client, _map_model_for_openrouter(model)
raise ValueError(
"No LLM key configured. Set OPENAI_API_KEY or OPENROUTER_API_KEY (universal fallback)."
)
async def transcribe_audio_whisper(
file_path: str,
model_size: str = "base",
language: str = "en"
) -> Union[str, TextContent]:
"""
Transcribe audio file using OpenAI Whisper (local).
Note: Requires whisper package installed.
Args:
file_path: Path to audio file
model_size: Whisper model size (tiny, base, small, medium, large)
language: Language code
Returns:
TextContent with transcription
"""
try:
path = validate_file_path(file_path)
logging.info(f"🎤 Transcribing audio: {path}")
try:
import whisper
# Load model
model = whisper.load_model(model_size)
# Transcribe
result = model.transcribe(str(path), language=language)
transcription = result["text"]
response_data = {
"file_name": path.name,
"file_type": path.suffix,
"model": model_size,
"language": language,
"transcription": transcription,
"word_count": len(transcription.split())
}
logging.info(f"✅ Transcribed: {len(transcription)} chars")
action_response = ActionResponse(
success=True,
message=response_data,
metadata={"file_path": str(path)}
)
except ImportError:
# Fallback: try using OpenAI API if available
import os
from openai import OpenAI
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ImportError("Whisper not installed and no OPENAI_API_KEY found")
client = OpenAI(api_key=api_key)
with open(path, "rb") as audio_file:
transcription = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language=language
)
response_data = {
"file_name": path.name,
"file_type": path.suffix,
"model": "whisper-1 (API)",
"language": language,
"transcription": transcription.text,
"word_count": len(transcription.text.split())
}
action_response = ActionResponse(
success=True,
message=response_data,
metadata={"file_path": str(path), "method": "openai_api"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Audio transcription failed: {str(e)}"
logging.error(f"Audio error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "audio_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_audio_metadata(
file_path: str
) -> Union[str, TextContent]:
"""
Extract audio file metadata using ffprobe.
Args:
file_path: Path to audio file
Returns:
TextContent with audio metadata
"""
try:
path = validate_file_path(file_path)
logging.info(f"🎵 Extracting audio metadata: {path}")
# Use ffprobe to get metadata
cmd = [
"ffprobe",
"-v", "quiet",
"-print_format", "json",
"-show_format",
"-show_streams",
str(path)
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
metadata = json.loads(result.stdout)
format_info = metadata.get("format", {})
streams = metadata.get("streams", [])
audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})
response_data = {
"file_name": path.name,
"file_size": path.stat().st_size,
"duration": float(format_info.get("duration", 0)),
"bit_rate": int(format_info.get("bit_rate", 0)),
"format": format_info.get("format_name"),
"codec": audio_stream.get("codec_name"),
"sample_rate": int(audio_stream.get("sample_rate", 0)) if audio_stream.get("sample_rate") else None,
"channels": int(audio_stream.get("channels", 0)) if audio_stream.get("channels") else None
}
logging.info(f"✅ Audio metadata extracted")
action_response = ActionResponse(
success=True,
message=response_data,
metadata={"file_path": str(path)}
)
else:
raise RuntimeError(f"ffprobe failed: {result.stderr}")
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Audio metadata extraction failed: {str(e)}"
logging.error(f"Audio metadata error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "audio_metadata_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_text_ocr(
image_path: str,
language: str = "eng"
) -> Union[str, TextContent]:
"""
Extract text from image using OCR.
Args:
image_path: Path to image file
language: OCR language (eng, chi_sim, etc.)
Returns:
TextContent with extracted text
"""
try:
path = validate_file_path(image_path)
logging.info(f"🔍 OCR extracting from image: {path}")
try:
import pytesseract
img = Image.open(path)
text = pytesseract.image_to_string(img, lang=language)
result = {
"file_name": path.name,
"image_size": img.size,
"extracted_text": text,
"text_length": len(text),
"word_count": len(text.split()),
"language": language,
"method": "pytesseract"
}
logging.info(f"✅ OCR extracted: {len(text)} chars")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
except ImportError:
# Fallback to a simpler method or error
raise ImportError("pytesseract not installed. Install with: pip install pytesseract")
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"OCR extraction failed: {str(e)}"
logging.error(f"OCR error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "ocr_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def analyze_image_ai(
image_path: str,
prompt: str = "Describe this image in detail"
) -> Union[str, TextContent]:
"""
Analyze image using AI (OpenAI Vision API).
Args:
image_path: Path to image file
prompt: Prompt for AI analysis
Returns:
TextContent with AI analysis
"""
try:
path = validate_file_path(image_path)
logging.info(f"🤖 AI analyzing image: {path}")
client, model = _make_vision_client()
# Encode image
with open(path, "rb") as img_file:
img_base64 = base64.b64encode(img_file.read()).decode('utf-8')
# Call Vision API
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{img_base64}"
}
}
]
}
],
max_tokens=500
)
analysis = response.choices[0].message.content
result = {
"file_name": path.name,
"prompt": prompt,
"analysis": analysis,
"model": model
}
logging.info(f"✅ AI analysis completed")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"AI image analysis failed: {str(e)}"
logging.error(f"AI analysis error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "ai_analysis_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_video_keyframes(
video_path: str,
num_frames: int = 10
) -> Union[str, TextContent]:
"""
Extract keyframes from video.
Args:
video_path: Path to video file
num_frames: Number of keyframes to extract
Returns:
TextContent with keyframe information
"""
try:
path = validate_file_path(video_path)
num_frames = max(1, num_frames)
logging.info(f"🎬 Extracting keyframes from video: {path}")
video = cv2.VideoCapture(str(path))
fps = video.get(cv2.CAP_PROP_FPS)
frame_count = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
duration = frame_count / fps if fps > 0 else 0
# Calculate frame interval
interval = max(1, frame_count // num_frames)
keyframes = []
frame_num = 0
while len(keyframes) < num_frames and video.isOpened():
ret, frame = video.read()
if not ret:
break
if frame_num % interval == 0:
timestamp = frame_num / fps if fps > 0 else 0
keyframes.append({
"frame_number": frame_num,
"timestamp": round(timestamp, 2),
"shape": frame.shape if frame is not None else None
})
frame_num += 1
video.release()
result = {
"file_name": path.name,
"duration": duration,
"total_frames": frame_count,
"fps": fps,
"keyframes_extracted": len(keyframes),
"keyframes": keyframes
}
logging.info(f"✅ Extracted {len(keyframes)} keyframes")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Keyframe extraction failed: {str(e)}"
logging.error(f"Video error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "video_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def analyze_video_ai(
video_path: str,
num_frames: int = 5,
prompt: str = "Analyze this video and describe what's happening"
) -> Union[str, TextContent]:
"""
Analyze video content using AI vision on keyframes.
Args:
video_path: Path to video file
num_frames: Number of frames to analyze
prompt: Analysis prompt for AI
Returns:
TextContent with AI analysis
"""
try:
path = validate_file_path(video_path)
num_frames = max(1, num_frames)
logging.info(f"🤖 AI analyzing video: {path}")
client, model = _make_vision_client()
# Extract keyframes
video = cv2.VideoCapture(str(path))
fps = video.get(cv2.CAP_PROP_FPS)
frame_count = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
interval = max(1, frame_count // num_frames)
# Extract and encode frames
frame_analyses = []
frame_num = 0
frames_analyzed = 0
while frames_analyzed < num_frames and video.isOpened():
ret, frame = video.read()
if not ret:
break
if frame_num % interval == 0:
# Encode frame
_, buffer = cv2.imencode('.jpg', frame)
img_base64 = base64.b64encode(buffer).decode('utf-8')
# Analyze with GPT-4 Vision
response = client.chat.completions.create(
model=model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": f"{prompt} (Frame {frames_analyzed + 1}/{num_frames})"},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img_base64}"}
}
]
}
],
max_tokens=300
)
timestamp = frame_num / fps if fps > 0 else 0
analysis = response.choices[0].message.content
frame_analyses.append({
"frame_number": frame_num,
"timestamp": round(timestamp, 2),
"analysis": analysis
})
frames_analyzed += 1
frame_num += 1
video.release()
# Generate overall summary
combined_analyses = "\n\n".join([f"Frame {i+1} (t={a['timestamp']}s): {a['analysis']}"
for i, a in enumerate(frame_analyses)])
result = {
"file_name": path.name,
"frames_analyzed": len(frame_analyses),
"analyses": frame_analyses,
"combined_analysis": combined_analyses
}
logging.info(f"✅ Analyzed {len(frame_analyses)} frames")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Video analysis failed: {str(e)}"
logging.error(f"Video analysis error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "video_analysis_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def trim_audio(
audio_path: str,
start_time: float,
duration: float | None = None,
output_path: str | None = None
) -> Union[str, TextContent]:
"""
Trim audio file to specified time range using ffmpeg.
Args:
audio_path: Path to audio file
start_time: Start time in seconds
duration: Duration in seconds (None for trim to end)
output_path: Output file path (None for auto-generate)
Returns:
TextContent with trimmed audio info
"""
try:
path = validate_file_path(audio_path)
logging.info(f"✂️ Trimming audio: {path}")
# Generate output path if not provided
if output_path is None:
output_path = str(path.parent / f"{path.stem}_trimmed{path.suffix}")
# Build ffmpeg command
cmd = ["ffmpeg", "-i", str(path), "-ss", str(start_time)]
if duration is not None:
cmd.extend(["-t", str(duration)])
cmd.extend(["-c", "copy", "-y", output_path])
# Execute ffmpeg
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg failed: {result.stderr}")
output_file = Path(output_path)
response_data = {
"input_file": str(path),
"output_file": str(output_file),
"start_time": start_time,
"duration": duration,
"file_size": output_file.stat().st_size if output_file.exists() else 0
}
logging.info(f"✅ Trimmed audio saved to: {output_file}")
action_response = ActionResponse(
success=True,
message=response_data,
metadata={"output_path": str(output_file)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Audio trim failed: {str(e)}"
logging.error(f"Audio trim error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "audio_trim_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_image_metadata(
image_path: str
) -> Union[str, TextContent]:
"""
Get detailed image metadata including EXIF data.
Args:
image_path: Path to image file
Returns:
TextContent with image metadata
"""
try:
path = validate_file_path(image_path)
logging.info(f"📷 Getting image metadata: {path}")
img = Image.open(path)
# Basic metadata
metadata = {
"file_name": path.name,
"format": img.format,
"mode": img.mode,
"size": img.size,
"width": img.width,
"height": img.height,
"file_size": path.stat().st_size
}
# Try to get EXIF data
try:
from PIL.ExifTags import TAGS
exif_data = {}
if hasattr(img, '_getexif') and img._getexif():
exif = img._getexif()
for tag_id, value in exif.items():
tag = TAGS.get(tag_id, tag_id)
exif_data[tag] = str(value)
if exif_data:
metadata["exif"] = exif_data
except Exception as e:
logging.debug(f"No EXIF data: {e}")
# Image info
if hasattr(img, 'info'):
metadata["info"] = dict(img.info)
result = {
"metadata": metadata,
"has_exif": "exif" in metadata
}
logging.info(f"✅ Image metadata extracted")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Metadata extraction failed: {str(e)}"
logging.error(f"Metadata error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "metadata_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/multimodal_tools.py¶
"""
Multimodal understanding tools: web, documents, images, and videos.
"""
import json
import logging
import os
import traceback
from pathlib import Path
from typing import Union
import base64
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from mcp.types import TextContent
from pydantic import Field
from base import ActionResponse, validate_file_path, download_file_from_url, is_url
load_dotenv()
async def read_webpage(
url: str,
extract_text: bool = True,
extract_links: bool = False
) -> Union[str, TextContent]:
"""
Read and extract content from a webpage.
Args:
url: URL of the webpage
extract_text: Whether to extract main text content
extract_links: Whether to extract all links
Returns:
TextContent with extracted webpage content
"""
try:
logging.info(f"📄 Reading webpage: {url}")
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
result = {
"url": url,
"title": soup.title.string if soup.title else "No title"
}
if extract_text:
# Remove script and style elements
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = ' '.join(chunk for chunk in chunks if chunk)
result["text"] = text[:5000] # Limit to first 5000 chars
result["text_length"] = len(text)
if extract_links:
links = []
for link in soup.find_all('a', href=True):
links.append({
"text": link.get_text().strip(),
"href": link['href']
})
result["links"] = links[:50] # Limit to first 50 links
logging.info(f"✅ Successfully extracted webpage content")
action_response = ActionResponse(
success=True,
message=result,
metadata={"url": url}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Webpage reading failed: {str(e)}"
logging.error(f"Webpage error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "webpage_error", "url": url}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def read_document(
file_path: str,
extract_images: bool = False
) -> Union[str, TextContent]:
"""
Read and extract content from documents (PDF, DOCX, PPTX).
Args:
file_path: Path to the document file (or URL)
extract_images: Whether to extract images from document
Returns:
TextContent with extracted document content
"""
try:
# Handle URL downloads
if is_url(file_path):
logging.info(f"📥 Downloading document from URL")
temp_path, _ = download_file_from_url(file_path)
file_path = temp_path
path = validate_file_path(file_path)
logging.info(f"📄 Reading document: {path}")
file_ext = path.suffix.lower()
# PDF extraction
if file_ext == '.pdf':
import PyPDF2
with open(path, 'rb') as file:
reader = PyPDF2.PdfReader(file)
text = ""
for page in reader.pages:
text += page.extract_text() + "\n"
result = {
"file_name": path.name,
"file_type": "pdf",
"page_count": len(reader.pages),
"text": text[:10000], # Limit size
"text_length": len(text)
}
# DOCX extraction
elif file_ext == '.docx':
from docx import Document
doc = Document(path)
text = "\n".join([para.text for para in doc.paragraphs])
result = {
"file_name": path.name,
"file_type": "docx",
"paragraph_count": len(doc.paragraphs),
"text": text[:10000],
"text_length": len(text)
}
# PPTX extraction
elif file_ext == '.pptx':
from pptx import Presentation
prs = Presentation(path)
text = ""
for slide in prs.slides:
for shape in slide.shapes:
if hasattr(shape, "text"):
text += shape.text + "\n"
result = {
"file_name": path.name,
"file_type": "pptx",
"slide_count": len(prs.slides),
"text": text[:10000],
"text_length": len(text)
}
else:
raise ValueError(f"Unsupported file type: {file_ext}")
logging.info(f"✅ Successfully extracted document content")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path), "file_type": file_ext}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Document reading failed: {str(e)}"
logging.error(f"Document error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "document_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def parse_image(
image_path: str,
use_llm: bool = True
) -> Union[str, TextContent]:
"""
Parse and understand image content.
Args:
image_path: Path to image file or URL
use_llm: Whether to use LLM for image understanding
Returns:
TextContent with image analysis
"""
try:
# Handle URL downloads
if is_url(image_path):
logging.info(f"📥 Downloading image from URL")
temp_path, _ = download_file_from_url(image_path)
image_path = temp_path
path = validate_file_path(image_path)
logging.info(f"🖼️ Parsing image: {path}")
from PIL import Image
img = Image.open(path)
result = {
"file_name": path.name,
"format": img.format,
"mode": img.mode,
"size": img.size,
"width": img.width,
"height": img.height
}
# If LLM analysis requested, encode image for vision API
if use_llm:
with open(path, 'rb') as img_file:
img_base64 = base64.b64encode(img_file.read()).decode('utf-8')
result["base64_data"] = img_base64[:100] + "..." # Truncated for display
result["note"] = "Full base64 data available for vision API analysis"
logging.info(f"✅ Successfully parsed image")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Image parsing failed: {str(e)}"
logging.error(f"Image error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "image_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def parse_video(
video_path: str,
extract_frames: bool = False,
frame_interval: int = 30
) -> Union[str, TextContent]:
"""
Parse and extract information from video files.
Args:
video_path: Path to video file or URL
extract_frames: Whether to extract sample frames
frame_interval: Extract one frame every N seconds
Returns:
TextContent with video metadata
"""
try:
# Handle URL downloads
if is_url(video_path):
logging.info(f"📥 Downloading video from URL")
temp_path, _ = download_file_from_url(video_path, max_size_mb=500)
video_path = temp_path
path = validate_file_path(video_path)
logging.info(f"🎥 Parsing video: {path}")
import cv2
video = cv2.VideoCapture(str(path))
fps = video.get(cv2.CAP_PROP_FPS)
frame_count = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
width = int(video.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
duration = frame_count / fps if fps > 0 else 0
result = {
"file_name": path.name,
"duration_seconds": duration,
"fps": fps,
"frame_count": frame_count,
"resolution": f"{width}x{height}",
"width": width,
"height": height
}
video.release()
logging.info(f"✅ Successfully parsed video metadata")
action_response = ActionResponse(
success=True,
message=result,
metadata={"file_path": str(path)}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Video parsing failed: {str(e)}"
logging.error(f"Video error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "video_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def download_youtube_video(
url: str,
output_dir: str = ".",
max_resolution: str = "720p"
) -> Union[str, TextContent]:
"""
Download YouTube video using yt-dlp.
Args:
url: YouTube video URL
output_dir: Directory to save video
max_resolution: Maximum resolution (360p, 480p, 720p, 1080p)
Returns:
TextContent with download result
"""
try:
logging.info(f"📥 Downloading YouTube video: {url}")
try:
import yt_dlp
output_template = Path(output_dir) / '%(title)s.%(ext)s'
ydl_opts = {
'format': f'best[height<={max_resolution[:-1]}]',
'outtmpl': str(output_template),
'quiet': False
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=True)
result = {
"title": info['title'],
"duration": info.get('duration'),
"output_dir": output_dir,
"resolution": max_resolution,
"video_id": info['id']
}
logging.info(f"✅ Downloaded: {info['title']}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"url": url}
)
except ImportError:
raise ImportError("yt-dlp not installed. Install with: pip install yt-dlp")
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"YouTube download failed: {str(e)}"
logging.error(f"YouTube download error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "youtube_download_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def extract_youtube_transcript(
video_id: str,
language_code: str = "en",
translate_to_language: str | None = None
) -> Union[str, TextContent]:
"""
Extract transcript from a YouTube video.
Args:
video_id: YouTube video ID or URL
language_code: Language code for the transcript (default: en)
translate_to_language: Translate transcript to this language if provided
Returns:
TextContent with transcript data
"""
try:
from youtube_transcript_api import YouTubeTranscriptApi
# Clean video_id if full URL was provided
if "youtube.com" in video_id or "youtu.be" in video_id:
if "?v=" in video_id:
video_id = video_id.split("?v=")[-1].split("&")[0]
elif "youtu.be/" in video_id:
video_id = video_id.split("youtu.be/")[-1].split("?")[0]
logging.info(f"📺 Extracting transcript for video ID: {video_id}")
# Get transcript using correct API
if translate_to_language:
transcript_list = YouTubeTranscriptApi().list(video_id)
try:
transcript = transcript_list.find_transcript([language_code])
except Exception:
# If specified language not found, get any available transcript
transcript = transcript_list.find_generated_transcript(["en"])
# Translate to target language
fetched_transcript = transcript.translate(translate_to_language).fetch()
transcript_data = fetched_transcript.snippets
else:
try:
# Use fetch method which returns FetchedTranscript
fetched_transcript = YouTubeTranscriptApi().fetch(
video_id,
languages=(language_code,)
)
transcript_data = fetched_transcript.snippets
except Exception:
# Fallback to English
fetched_transcript = YouTubeTranscriptApi().fetch(video_id, languages=("en",))
transcript_data = fetched_transcript.snippets
# Format transcript
formatted_transcript = []
for entry in transcript_data:
# Access as object attributes, not dictionary
start_time = entry.start if hasattr(entry, 'start') else entry.get('start', 0)
text = entry.text if hasattr(entry, 'text') else entry.get('text', '')
minutes, seconds = divmod(int(start_time), 60)
timestamp = f"{minutes:02d}:{seconds:02d}"
formatted_transcript.append({
"timestamp": timestamp,
"text": text
})
# Create full text version
full_text = " ".join([
entry.text if hasattr(entry, 'text') else entry.get('text', '')
for entry in transcript_data
])
result = {
"video_id": video_id,
"language": translate_to_language if translate_to_language else language_code,
"transcript": formatted_transcript[:100], # Limit to first 100 entries
"total_entries": len(transcript_data),
"full_text": full_text[:5000], # Limit full text to 5000 chars
"full_text_length": len(full_text)
}
logging.info(f"✅ Successfully extracted transcript ({len(transcript_data)} entries)")
action_response = ActionResponse(
success=True,
message=result,
metadata={
"video_id": video_id,
"language": language_code,
"translated": translate_to_language is not None
}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"YouTube transcript extraction failed: {str(e)}"
logging.error(f"YouTube error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "youtube_error", "video_id": video_id}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/private_data_tools.py¶
"""
Private data source tools: Google Calendar, Notion.
"""
import json
import logging
import os
import traceback
from datetime import datetime, timedelta
from typing import Union
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse
load_dotenv()
async def get_calendar_events(
start_date: str | None = None,
end_date: str | None = None,
calendar_id: str = "primary",
max_results: int = 10
) -> Union[str, TextContent]:
"""
Get events from Google Calendar.
Args:
start_date: Start date (ISO format, defaults to today)
end_date: End date (ISO format, defaults to 7 days from now)
calendar_id: Calendar ID (default: primary)
max_results: Maximum number of events to return
Returns:
TextContent with calendar events
"""
try:
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from google.auth.transport.requests import Request
import pickle
logging.info(f"📅 Getting calendar events")
# Token file path
token_path = os.path.expanduser("~/.perception-tools/google_token.pickle")
if not os.path.exists(token_path):
action_response = ActionResponse(
success=False,
message="Google Calendar not configured. Please run setup to authenticate.",
metadata={"error_type": "missing_credentials"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
# Load credentials
with open(token_path, 'rb') as token:
creds = pickle.load(token)
# Refresh if expired
if creds.expired and creds.refresh_token:
creds.refresh(Request())
service = build('calendar', 'v3', credentials=creds)
# Set default date range if not provided
if not start_date:
start_date = datetime.utcnow().isoformat() + 'Z'
if not end_date:
end_dt = datetime.utcnow() + timedelta(days=7)
end_date = end_dt.isoformat() + 'Z'
# Query calendar
events_result = service.events().list(
calendarId=calendar_id,
timeMin=start_date,
timeMax=end_date,
maxResults=max_results,
singleEvents=True,
orderBy='startTime'
).execute()
events = events_result.get('items', [])
formatted_events = []
for event in events:
start = event['start'].get('dateTime', event['start'].get('date'))
end = event['end'].get('dateTime', event['end'].get('date'))
formatted_events.append({
"id": event['id'],
"summary": event.get('summary', 'No title'),
"start": start,
"end": end,
"location": event.get('location'),
"description": event.get('description'),
"attendees": [a.get('email') for a in event.get('attendees', [])]
})
logging.info(f"✅ Found {len(formatted_events)} calendar events")
action_response = ActionResponse(
success=True,
message={
"events": formatted_events,
"count": len(formatted_events),
"calendar_id": calendar_id
},
metadata={"start_date": start_date, "end_date": end_date}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except ImportError:
error_msg = "Google Calendar libraries not installed. Install with: pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client"
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "missing_library"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Calendar query failed: {str(e)}"
logging.error(f"Calendar error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "calendar_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_notion(
query: str,
database_id: str | None = None,
page_size: int = 10
) -> Union[str, TextContent]:
"""
Search Notion workspace or specific database.
Args:
query: Search query
database_id: Optional specific database ID
page_size: Number of results per page
Returns:
TextContent with Notion search results
"""
try:
from notion_client import Client
logging.info(f"📝 Searching Notion for: {query}")
api_key = os.getenv("NOTION_API_KEY")
if not api_key:
action_response = ActionResponse(
success=False,
message="Notion API key not configured. Set NOTION_API_KEY.",
metadata={"error_type": "missing_credentials"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
notion = Client(auth=api_key)
if database_id:
# Search specific database
response = notion.databases.query(
database_id=database_id,
filter={
"property": "Name",
"rich_text": {
"contains": query
}
},
page_size=page_size
)
else:
# Search entire workspace
response = notion.search(
query=query,
page_size=page_size
)
results = []
for item in response.get("results", []):
result_data = {
"id": item["id"],
"type": item["object"],
"url": item.get("url"),
"created_time": item.get("created_time"),
"last_edited_time": item.get("last_edited_time")
}
# Extract title/name
if "properties" in item:
for prop_name, prop_value in item["properties"].items():
if prop_value.get("type") == "title" and prop_value.get("title"):
title_parts = [t.get("plain_text", "") for t in prop_value["title"]]
result_data["title"] = "".join(title_parts)
break
results.append(result_data)
logging.info(f"✅ Found {len(results)} Notion items")
action_response = ActionResponse(
success=True,
message={
"query": query,
"results": results,
"count": len(results)
},
metadata={"database_id": database_id}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except ImportError:
error_msg = "Notion SDK not installed. Install with: pip install notion-client"
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "missing_library"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Notion search failed: {str(e)}"
logging.error(f"Notion error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "notion_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/pubchem_tools.py¶
"""
PubChem chemical compound data tools.
Based on AWorld MCP server implementation.
"""
import json
import logging
import time
import traceback
from typing import Union, Literal
from urllib.parse import quote
import requests
from dotenv import load_dotenv
from mcp.types import TextContent
from pydantic import BaseModel, Field
from base import ActionResponse
load_dotenv()
class CompoundData(BaseModel):
"""Structured compound data from PubChem."""
cid: int | None = None
name: str | None = None
molecular_formula: str | None = None
molecular_weight: float | None = None
smiles: str | None = None
inchi: str | None = None
synonyms: list[str] = []
class PubChemMetadata(BaseModel):
"""Metadata for PubChem operation results."""
query_type: str
query_value: str
api_endpoint: str
response_time: float
total_results: int | None = None
rate_limit_delay: float | None = None
error_type: str | None = None
class PubChemClient:
"""PubChem API client with rate limiting."""
def __init__(self):
self.base_url = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"
self.request_delay = 0.2 # 200ms delay to stay under 5 req/sec limit
self.last_request_time = 0.0
self.timeout = 30
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "PerceptionToolsMCP/1.0",
"Accept": "application/json"
})
def _rate_limit(self) -> float:
"""Enforce rate limiting to comply with PubChem usage policy."""
current_time = time.time()
time_since_last = current_time - self.last_request_time
if time_since_last < self.request_delay:
delay = self.request_delay - time_since_last
time.sleep(delay)
self.last_request_time = time.time()
return delay
self.last_request_time = current_time
return 0.0
def make_request(self, url: str, params: dict = None, max_retries: int = 3) -> tuple[dict | None, float]:
"""Make a rate-limited request to PubChem API with retry for async operations."""
self._rate_limit()
start_time = time.time()
try:
response = self.session.get(url, params=params, timeout=self.timeout)
response_time = time.time() - start_time
if response.status_code == 200:
return response.json(), response_time
elif response.status_code == 202:
# Async operation - wait and retry
if max_retries > 0:
logging.info(f"PubChem async operation, waiting 2s before retry...")
time.sleep(2)
return self.make_request(url, params, max_retries - 1)
else:
raise requests.RequestException("PubChem async operation timeout after retries")
elif response.status_code == 503:
raise requests.RequestException("PubChem service temporarily unavailable (503)")
else:
raise requests.RequestException(f"HTTP {response.status_code}: {response.text}")
except requests.Timeout:
response_time = time.time() - start_time
raise requests.RequestException(f"Request timeout after {self.timeout}s")
except requests.RequestException:
response_time = time.time() - start_time
raise
# Global client instance
_client = None
def get_client() -> PubChemClient:
"""Get or create the global PubChem client."""
global _client
if _client is None:
_client = PubChemClient()
return _client
async def search_compounds(
query: str,
search_type: Literal["name", "cid", "smiles", "inchi", "formula"] = "name",
max_results: int = 10
) -> Union[str, TextContent]:
"""
Search for chemical compounds in PubChem database.
Args:
query: Search term or identifier
search_type: Type of search (name, cid, smiles, inchi, formula)
max_results: Maximum number of results (1-100)
Returns:
TextContent with compound search results
"""
try:
if not query or not query.strip():
raise ValueError("Search query is required")
max_results = max(1, min(max_results, 100))
logging.info(f"🔬 Searching PubChem for: {query} (type: {search_type})")
client = get_client()
# Build API URL based on search type
if search_type == "cid":
url = f"{client.base_url}/compound/cid/{quote(str(query))}/property/Title,MolecularFormula,MolecularWeight,CanonicalSMILES,InChI/JSON"
elif search_type == "name":
url = f"{client.base_url}/compound/name/{quote(query)}/property/Title,MolecularFormula,MolecularWeight,CanonicalSMILES,InChI/JSON"
elif search_type == "smiles":
url = f"{client.base_url}/compound/smiles/{quote(query)}/property/Title,MolecularFormula,MolecularWeight,CanonicalSMILES,InChI/JSON"
elif search_type == "inchi":
url = f"{client.base_url}/compound/inchi/{quote(query)}/property/Title,MolecularFormula,MolecularWeight,CanonicalSMILES,InChI/JSON"
elif search_type == "formula":
url = f"{client.base_url}/compound/formula/{quote(query)}/property/Title,MolecularFormula,MolecularWeight,CanonicalSMILES,InChI/JSON"
else:
raise ValueError(f"Unsupported search type: {search_type}")
# Make API request
data, response_time = client.make_request(url)
# Parse results
compounds = []
if data and "PropertyTable" in data and "Properties" in data["PropertyTable"]:
properties_list = data["PropertyTable"]["Properties"][:max_results]
for prop in properties_list:
compound = CompoundData(
cid=prop.get("CID"),
name=prop.get("Title"),
molecular_formula=prop.get("MolecularFormula"),
molecular_weight=prop.get("MolecularWeight"),
smiles=prop.get("CanonicalSMILES"),
inchi=prop.get("InChI")
)
compounds.append(compound)
# Format results
result = {
"query": query,
"search_type": search_type,
"compounds": [c.model_dump() for c in compounds],
"count": len(compounds)
}
metadata = PubChemMetadata(
query_type=search_type,
query_value=query,
api_endpoint=url,
response_time=response_time,
total_results=len(compounds)
)
logging.info(f"✅ Found {len(compounds)} compounds ({response_time:.2f}s)")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except ValueError as e:
error_msg = f"Invalid input: {str(e)}"
logging.error(f"PubChem search error: {error_msg}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "invalid_input"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Search failed: {str(e)}"
logging.error(f"PubChem error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "api_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_compound_properties(
cid: int,
properties: list[str] | None = None
) -> Union[str, TextContent]:
"""
Retrieve detailed chemical properties for a PubChem compound.
Args:
cid: PubChem Compound ID
properties: List of property names (e.g., MolecularWeight, XLogP)
Returns:
TextContent with compound properties
"""
try:
if not cid or cid <= 0:
raise ValueError("Valid PubChem CID is required")
if not properties:
properties = [
"MolecularWeight", "MolecularFormula", "CanonicalSMILES",
"InChI", "XLogP", "TPSA", "HBondDonorCount", "HBondAcceptorCount"
]
logging.info(f"🔬 Getting properties for CID: {cid}")
client = get_client()
props_str = ",".join(properties)
url = f"{client.base_url}/compound/cid/{cid}/property/{props_str}/JSON"
data, response_time = client.make_request(url)
compound_props = {}
if data and "PropertyTable" in data and "Properties" in data["PropertyTable"]:
props_data = data["PropertyTable"]["Properties"][0]
compound_props = {k: v for k, v in props_data.items() if k != "CID"}
result = {
"cid": cid,
"properties": compound_props
}
metadata = PubChemMetadata(
query_type="properties",
query_value=str(cid),
api_endpoint=url,
response_time=response_time,
total_results=len(compound_props)
)
logging.info(f"✅ Retrieved {len(compound_props)} properties")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Property retrieval failed: {str(e)}"
logging.error(f"PubChem error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "api_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_compound_synonyms(
cid: int,
max_synonyms: int = 20
) -> Union[str, TextContent]:
"""
Retrieve synonyms for a PubChem compound.
Args:
cid: PubChem Compound ID
max_synonyms: Maximum number of synonyms (1-100)
Returns:
TextContent with compound synonyms
"""
try:
if not cid or cid <= 0:
raise ValueError("Valid PubChem CID is required")
max_synonyms = max(1, min(max_synonyms, 100))
logging.info(f"🔬 Getting synonyms for CID: {cid}")
client = get_client()
url = f"{client.base_url}/compound/cid/{cid}/synonyms/JSON"
data, response_time = client.make_request(url)
synonyms = []
if data and "InformationList" in data and "Information" in data["InformationList"]:
info_list = data["InformationList"]["Information"]
if info_list and "Synonym" in info_list[0]:
synonyms = info_list[0]["Synonym"][:max_synonyms]
result = {
"cid": cid,
"synonyms": synonyms,
"count": len(synonyms)
}
metadata = PubChemMetadata(
query_type="synonyms",
query_value=str(cid),
api_endpoint=url,
response_time=response_time,
total_results=len(synonyms)
)
logging.info(f"✅ Retrieved {len(synonyms)} synonyms")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Synonym retrieval failed: {str(e)}"
logging.error(f"PubChem error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "api_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_similar_compounds(
cid: int,
similarity_threshold: float = 0.9,
max_results: int = 10
) -> Union[str, TextContent]:
"""
Find structurally similar compounds.
Args:
cid: Reference compound CID
similarity_threshold: Minimum similarity (0.0-1.0)
max_results: Maximum results (1-50)
Returns:
TextContent with similar compounds
"""
try:
if not cid or cid <= 0:
raise ValueError("Valid PubChem CID is required")
similarity_threshold = max(0.0, min(similarity_threshold, 1.0))
max_results = max(1, min(max_results, 50))
logging.info(f"🔬 Searching similar compounds to CID: {cid}")
client = get_client()
threshold_percent = int(similarity_threshold * 100)
url = f"{client.base_url}/compound/fastsimilarity_2d/cid/{cid}/property/Title,MolecularFormula,MolecularWeight/JSON"
params = {
"Threshold": threshold_percent,
"MaxRecords": max_results
}
data, response_time = client.make_request(url, params)
similar_compounds = []
if data and "PropertyTable" in data and "Properties" in data["PropertyTable"]:
properties_list = data["PropertyTable"]["Properties"]
for prop in properties_list:
if prop.get("CID") != cid:
compound = CompoundData(
cid=prop.get("CID"),
name=prop.get("Title"),
molecular_formula=prop.get("MolecularFormula"),
molecular_weight=prop.get("MolecularWeight")
)
similar_compounds.append(compound)
result = {
"reference_cid": cid,
"similarity_threshold": similarity_threshold,
"similar_compounds": [c.model_dump() for c in similar_compounds],
"count": len(similar_compounds)
}
metadata = PubChemMetadata(
query_type="similarity",
query_value=str(cid),
api_endpoint=url,
response_time=response_time,
total_results=len(similar_compounds)
)
logging.info(f"✅ Found {len(similar_compounds)} similar compounds")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Similarity search failed: {str(e)}"
logging.error(f"PubChem error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "api_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/public_data_tools.py¶
"""
Public data source tools: weather, stocks, currency, Wiki, ArXiv, Wayback Machine.
"""
import json
import logging
import os
import time
import traceback
from datetime import datetime
from typing import Union
import requests
from dotenv import load_dotenv
from mcp.types import TextContent
from pydantic import BaseModel, Field
import wikipedia
from base import ActionResponse
load_dotenv()
async def get_weather(
location: str,
latitude: float | None = None,
longitude: float | None = None
) -> Union[str, TextContent]:
"""
Get current weather information for a location using Open-Meteo API.
Args:
location: City name for display purposes
latitude: Latitude coordinate (if not provided, will try to geocode location)
longitude: Longitude coordinate (if not provided, will try to geocode location)
Returns:
TextContent with weather data
"""
try:
logging.info(f"🌤️ Getting weather for: {location}")
# If coordinates not provided, try to geocode the location
if latitude is None or longitude is None:
# Use Open-Meteo's geocoding API
geocode_url = "https://geocoding-api.open-meteo.com/v1/search"
geocode_params = {
"name": location,
"count": 1,
"language": "en",
"format": "json"
}
geocode_response = requests.get(geocode_url, params=geocode_params, timeout=10)
geocode_response.raise_for_status()
geocode_data = geocode_response.json()
if not geocode_data.get("results"):
raise ValueError(f"Location not found: {location}")
first_result = geocode_data["results"][0]
latitude = first_result["latitude"]
longitude = first_result["longitude"]
location = first_result.get("name", location)
country = first_result.get("country", "")
else:
country = ""
# Get weather data from Open-Meteo
weather_url = "https://api.open-meteo.com/v1/forecast"
weather_params = {
"latitude": latitude,
"longitude": longitude,
"current": "temperature_2m,relative_humidity_2m,apparent_temperature,precipitation,weather_code,wind_speed_10m,wind_direction_10m",
"timezone": "auto"
}
response = requests.get(weather_url, params=weather_params, timeout=10)
response.raise_for_status()
data = response.json()
current = data["current"]
# Map weather codes to descriptions
# Based on WMO Weather interpretation codes
weather_codes = {
0: "Clear sky",
1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast",
45: "Foggy", 48: "Depositing rime fog",
51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle",
61: "Slight rain", 63: "Moderate rain", 65: "Heavy rain",
71: "Slight snow", 73: "Moderate snow", 75: "Heavy snow",
77: "Snow grains",
80: "Slight rain showers", 81: "Moderate rain showers", 82: "Violent rain showers",
85: "Slight snow showers", 86: "Heavy snow showers",
95: "Thunderstorm", 96: "Thunderstorm with slight hail", 99: "Thunderstorm with heavy hail"
}
weather_code = current["weather_code"]
description = weather_codes.get(weather_code, "Unknown")
result = {
"location": location,
"country": country,
"latitude": latitude,
"longitude": longitude,
"temperature": current["temperature_2m"],
"feels_like": current["apparent_temperature"],
"humidity": current["relative_humidity_2m"],
"precipitation": current["precipitation"],
"weather_code": weather_code,
"description": description,
"wind_speed": current["wind_speed_10m"],
"wind_direction": current["wind_direction_10m"],
"units": "metric",
"timestamp": current["time"],
"provider": "Open-Meteo"
}
logging.info(f"✅ Weather: {result['temperature']}°C - {result['description']}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"location": location, "provider": "Open-Meteo", "api_key_required": False}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Weather query failed: {str(e)}"
logging.error(f"Weather error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "weather_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_stock_price(
symbol: str,
interval: str = "1d"
) -> Union[str, TextContent]:
"""
Get stock price information.
Args:
symbol: Stock ticker symbol (e.g., AAPL, TSLA)
interval: Data interval (1d, 1h, etc.)
Returns:
TextContent with stock data
"""
try:
logging.info(f"📈 Getting stock price for: {symbol}")
# Using Yahoo Finance API (free, no key required)
url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}"
params = {
"interval": interval,
"range": "1d"
}
headers = {
"User-Agent": "Mozilla/5.0"
}
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
if "chart" in data and "result" in data["chart"] and data["chart"]["result"]:
quote = data["chart"]["result"][0]["meta"]
result = {
"symbol": symbol,
"currency": quote.get("currency", "USD"),
"current_price": quote.get("regularMarketPrice"),
"previous_close": quote.get("previousClose"),
"open": quote.get("regularMarketOpen"),
"day_high": quote.get("regularMarketDayHigh"),
"day_low": quote.get("regularMarketDayLow"),
"volume": quote.get("regularMarketVolume"),
"exchange": quote.get("exchangeName")
}
logging.info(f"✅ Stock price: ${result['current_price']}")
else:
raise ValueError(f"Invalid response for symbol: {symbol}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"symbol": symbol}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Stock query failed: {str(e)}"
logging.error(f"Stock error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "stock_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def convert_currency(
amount: float,
from_currency: str,
to_currency: str
) -> Union[str, TextContent]:
"""
Convert between currencies.
Args:
amount: Amount to convert
from_currency: Source currency code (e.g., USD)
to_currency: Target currency code (e.g., EUR)
Returns:
TextContent with conversion result
"""
try:
logging.info(f"💱 Converting {amount} {from_currency} to {to_currency}")
# Using free exchange rate API
url = f"https://api.exchangerate-api.com/v4/latest/{from_currency}"
response = requests.get(url, timeout=10)
response.raise_for_status()
data = response.json()
if to_currency not in data["rates"]:
raise ValueError(f"Currency not found: {to_currency}")
rate = data["rates"][to_currency]
converted_amount = amount * rate
result = {
"amount": amount,
"from_currency": from_currency,
"to_currency": to_currency,
"exchange_rate": rate,
"converted_amount": converted_amount,
"timestamp": data.get("date", datetime.now().isoformat())
}
logging.info(f"✅ {amount} {from_currency} = {converted_amount:.2f} {to_currency}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"rate": rate}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Currency conversion failed: {str(e)}"
logging.error(f"Currency error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "currency_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_wikipedia(
query: str,
language: str = "en",
sentences: int = 5
) -> Union[str, TextContent]:
"""
Search Wikipedia and get article summary.
Args:
query: Search query
language: Wikipedia language (en, zh, etc.)
sentences: Number of sentences in summary
Returns:
TextContent with Wikipedia article
"""
try:
wikipedia.set_lang(language)
logging.info(f"📚 Searching Wikipedia for: {query}")
# Search for pages
search_results = wikipedia.search(query, results=3)
if not search_results:
raise ValueError(f"No Wikipedia articles found for: {query}")
# Get the first result's page
page = wikipedia.page(search_results[0], auto_suggest=False)
summary = wikipedia.summary(search_results[0], sentences=sentences, auto_suggest=False)
result = {
"title": page.title,
"url": page.url,
"summary": summary,
"language": language,
"search_results": search_results
}
logging.info(f"✅ Found Wikipedia article: {page.title}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"query": query, "language": language}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Wikipedia search failed: {str(e)}"
logging.error(f"Wikipedia error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "wikipedia_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_arxiv(
query: str,
max_results: int = 5,
sort_by: str = "relevance"
) -> Union[str, TextContent]:
"""
Search ArXiv for academic papers.
Args:
query: Search query
max_results: Maximum number of results
sort_by: Sort method (relevance, lastUpdatedDate, submittedDate)
Returns:
TextContent with ArXiv papers
"""
try:
import arxiv
logging.info(f"🔬 Searching ArXiv for: {query}")
# Map sort_by to arxiv.SortCriterion
sort_map = {
"relevance": arxiv.SortCriterion.Relevance,
"lastUpdatedDate": arxiv.SortCriterion.LastUpdatedDate,
"submittedDate": arxiv.SortCriterion.SubmittedDate
}
sort_criterion = sort_map.get(sort_by, arxiv.SortCriterion.Relevance)
search = arxiv.Search(
query=query,
max_results=max_results,
sort_by=sort_criterion
)
papers = []
for result in search.results():
papers.append({
"title": result.title,
"authors": [author.name for author in result.authors],
"summary": result.summary[:500] + "...",
"published": result.published.isoformat(),
"url": result.entry_id,
"pdf_url": result.pdf_url,
"categories": result.categories
})
logging.info(f"✅ Found {len(papers)} papers")
action_response = ActionResponse(
success=True,
message={
"query": query,
"papers": papers,
"count": len(papers)
},
metadata={"query": query, "max_results": max_results}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"ArXiv search failed: {str(e)}"
logging.error(f"ArXiv error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "arxiv_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_wayback(
url: str,
year: int | None = None,
limit: int = 10
) -> Union[str, TextContent]:
"""
Search Wayback Machine for archived versions of a URL.
Args:
url: URL to search for
year: Optional year to filter results
limit: Maximum number of snapshots to return
Returns:
TextContent with archived snapshots
"""
try:
logging.info(f"🕰️ Searching Wayback Machine for: {url}")
# CDX API endpoint
cdx_url = "http://web.archive.org/cdx/search/cdx"
params = {
"url": url,
"output": "json",
"limit": limit,
"fl": "timestamp,original,statuscode,mimetype"
}
if year:
params["from"] = f"{year}0101"
params["to"] = f"{year}1231"
response = requests.get(cdx_url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
# First row is headers
if len(data) <= 1:
raise ValueError(f"No archived snapshots found for: {url}")
headers = data[0]
snapshots = []
for row in data[1:]:
snapshot = dict(zip(headers, row))
# Convert timestamp to readable format
ts = snapshot["timestamp"]
dt = datetime.strptime(ts, "%Y%m%d%H%M%S")
snapshots.append({
"timestamp": dt.isoformat(),
"url": f"https://web.archive.org/web/{ts}/{snapshot['original']}",
"status_code": snapshot.get("statuscode"),
"mime_type": snapshot.get("mimetype")
})
logging.info(f"✅ Found {len(snapshots)} archived snapshots")
action_response = ActionResponse(
success=True,
message={
"url": url,
"snapshots": snapshots,
"count": len(snapshots)
},
metadata={"url": url, "year": year}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Wayback Machine search failed: {str(e)}"
logging.error(f"Wayback error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "wayback_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_crypto_price(
symbol: str,
vs_currency: str = "usd"
) -> Union[str, TextContent]:
"""
Get cryptocurrency price information using CoinGecko API (free, no API key required).
Args:
symbol: Cryptocurrency symbol or ID (e.g., bitcoin, ethereum, btc, eth)
vs_currency: Target currency (usd, eur, gbp, etc.)
Returns:
TextContent with cryptocurrency data
"""
try:
logging.info(f"💰 Getting crypto price for: {symbol}")
# CoinGecko free API
# First, try to get the coin ID from symbol
symbol_lower = symbol.lower()
# Map common symbols to CoinGecko IDs
symbol_map = {
"btc": "bitcoin",
"eth": "ethereum",
"usdt": "tether",
"bnb": "binancecoin",
"sol": "solana",
"xrp": "ripple",
"usdc": "usd-coin",
"ada": "cardano",
"doge": "dogecoin",
"trx": "tron",
"dot": "polkadot",
"matic": "matic-network",
"dai": "dai",
"shib": "shiba-inu",
"avax": "avalanche-2"
}
# Use mapped ID or try the symbol directly
coin_id = symbol_map.get(symbol_lower, symbol_lower)
# Get price data from CoinGecko
url = "https://api.coingecko.com/api/v3/simple/price"
params = {
"ids": coin_id,
"vs_currencies": vs_currency,
"include_market_cap": "true",
"include_24hr_vol": "true",
"include_24hr_change": "true",
"include_last_updated_at": "true"
}
headers = {
"User-Agent": "Mozilla/5.0"
}
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
if not data or coin_id not in data:
raise ValueError(f"Cryptocurrency not found: {symbol}")
coin_data = data[coin_id]
result = {
"symbol": symbol.upper(),
"coin_id": coin_id,
"currency": vs_currency.upper(),
"current_price": coin_data.get(vs_currency),
"market_cap": coin_data.get(f"{vs_currency}_market_cap"),
"volume_24h": coin_data.get(f"{vs_currency}_24h_vol"),
"price_change_24h_percent": coin_data.get(f"{vs_currency}_24h_change"),
"last_updated": datetime.fromtimestamp(coin_data.get("last_updated_at", 0)).isoformat() if coin_data.get("last_updated_at") else None,
"provider": "CoinGecko"
}
logging.info(f"✅ Crypto price: {result['current_price']} {vs_currency.upper()}")
action_response = ActionResponse(
success=True,
message=result,
metadata={"symbol": symbol, "provider": "CoinGecko", "api_key_required": False}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Crypto price query failed: {str(e)}"
logging.error(f"Crypto error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "crypto_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_location(
query: str,
limit: int = 5,
country_code: str | None = None
) -> Union[str, TextContent]:
"""
Search for locations using Nominatim (OpenStreetMap) API (free, no API key required).
Args:
query: Location query (e.g., "Eiffel Tower", "New York", "coffee shop near me")
limit: Maximum number of results (1-50)
country_code: Optional country code filter (e.g., "us", "gb", "fr")
Returns:
TextContent with location search results
"""
try:
logging.info(f"📍 Searching location: {query}")
# Nominatim API (OpenStreetMap)
url = "https://nominatim.openstreetmap.org/search"
params = {
"q": query,
"format": "json",
"limit": min(limit, 50),
"addressdetails": 1,
"extratags": 1
}
if country_code:
params["countrycodes"] = country_code.lower()
headers = {
"User-Agent": "PerceptionToolsMCP/1.0"
}
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
if not data:
raise ValueError(f"No locations found for: {query}")
locations = []
for item in data:
address = item.get("address", {})
locations.append({
"display_name": item.get("display_name"),
"latitude": float(item.get("lat")),
"longitude": float(item.get("lon")),
"type": item.get("type"),
"category": item.get("class"),
"address": {
"country": address.get("country"),
"country_code": address.get("country_code"),
"state": address.get("state"),
"city": address.get("city") or address.get("town") or address.get("village"),
"postcode": address.get("postcode"),
"road": address.get("road")
},
"importance": item.get("importance"),
"osm_id": item.get("osm_id"),
"osm_type": item.get("osm_type")
})
logging.info(f"✅ Found {len(locations)} locations")
action_response = ActionResponse(
success=True,
message={
"query": query,
"locations": locations,
"count": len(locations)
},
metadata={"provider": "Nominatim (OpenStreetMap)", "api_key_required": False}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Location search failed: {str(e)}"
logging.error(f"Location search error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "location_search_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_poi(
query: str,
latitude: float,
longitude: float,
radius: int = 1000,
limit: int = 10
) -> Union[str, TextContent]:
"""
Search for Points of Interest (POI) near a location using Overpass API (OpenStreetMap).
Free, no API key required.
Args:
query: Type of POI (e.g., "restaurant", "cafe", "hospital", "atm", "hotel")
latitude: Center latitude
longitude: Center longitude
radius: Search radius in meters (default: 1000)
limit: Maximum number of results (default: 10)
Returns:
TextContent with POI search results
"""
try:
logging.info(f"🔍 Searching POIs: {query} near ({latitude}, {longitude})")
# Overpass API query
# Search for amenities, shops, tourism, etc.
overpass_query = f"""
[out:json][timeout:10];
(
node["amenity"~"{query}",i](around:{radius},{latitude},{longitude});
node["shop"~"{query}",i](around:{radius},{latitude},{longitude});
node["tourism"~"{query}",i](around:{radius},{latitude},{longitude});
node["name"~"{query}",i](around:{radius},{latitude},{longitude});
);
out body {limit};
"""
url = "https://overpass-api.de/api/interpreter"
response = requests.post(url, data={"data": overpass_query}, timeout=30)
response.raise_for_status()
data = response.json()
elements = data.get("elements", [])
if not elements:
raise ValueError(f"No POIs found for '{query}' near the specified location")
pois = []
for element in elements[:limit]:
tags = element.get("tags", {})
pois.append({
"name": tags.get("name", "Unnamed"),
"type": tags.get("amenity") or tags.get("shop") or tags.get("tourism") or "unknown",
"latitude": element.get("lat"),
"longitude": element.get("lon"),
"address": tags.get("addr:street"),
"city": tags.get("addr:city"),
"postcode": tags.get("addr:postcode"),
"phone": tags.get("phone"),
"website": tags.get("website"),
"opening_hours": tags.get("opening_hours"),
"cuisine": tags.get("cuisine"),
"osm_id": element.get("id"),
"osm_type": element.get("type")
})
logging.info(f"✅ Found {len(pois)} POIs")
action_response = ActionResponse(
success=True,
message={
"query": query,
"center": {"latitude": latitude, "longitude": longitude},
"radius_meters": radius,
"pois": pois,
"count": len(pois)
},
metadata={"provider": "Overpass API (OpenStreetMap)", "api_key_required": False}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"POI search failed: {str(e)}"
logging.error(f"POI search error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "poi_search_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/search_tools.py¶
"""
Search tools: knowledge base, web search, and file download.
"""
import json
import logging
import os
import time
import traceback
from pathlib import Path
from typing import Union
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from mcp.types import TextContent
from pydantic import BaseModel, Field
from base import ActionResponse, is_url, download_file_from_url
load_dotenv()
class SearchResult(BaseModel):
"""Individual search result with structured data."""
id: str
title: str
url: str
snippet: str
source: str
class SearchMetadata(BaseModel):
"""Metadata for search operations."""
query: str
search_engine: str
total_results: int
search_time: float | None = None
language: str = "en"
country: str = "us"
async def search_web(
query: str,
num_results: int = 5,
region: str = "wt-wt"
) -> Union[str, TextContent]:
"""
Search the web using DuckDuckGo (free, no API key required).
Args:
query: The search query string
num_results: Number of results to return (1-10)
region: Region code (e.g., 'us-en', 'uk-en', 'wt-wt' for worldwide)
Returns:
TextContent with search results
"""
try:
if not query or not query.strip():
raise ValueError("Search query cannot be empty")
validated_num_results = max(1, min(num_results, 10))
logging.info(f"🔍 Searching for: '{query}'")
start_time = time.time()
# Use DuckDuckGo HTML version for scraping
url = "https://html.duckduckgo.com/html/"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
data = {
"q": query.strip(),
"kl": region
}
response = requests.post(url, data=data, headers=headers, timeout=15)
response.raise_for_status()
search_time = time.time() - start_time
# Parse HTML results
soup = BeautifulSoup(response.text, 'html.parser')
result_divs = soup.find_all('div', class_='result')
search_results = []
for i, result_div in enumerate(result_divs[:validated_num_results]):
try:
# Extract title and URL
title_tag = result_div.find('a', class_='result__a')
if not title_tag:
continue
title = title_tag.get_text(strip=True)
url_link = title_tag.get('href', '')
# Extract snippet
snippet_tag = result_div.find('a', class_='result__snippet')
snippet = snippet_tag.get_text(strip=True) if snippet_tag else ""
result = SearchResult(
id=f"ddg-{i}",
title=title,
url=url_link,
snippet=snippet,
source="duckduckgo"
)
search_results.append(result)
except Exception as e:
logging.warning(f"Error parsing search result {i}: {e}")
continue
metadata = SearchMetadata(
query=query,
search_engine="duckduckgo",
total_results=len(search_results),
search_time=search_time,
language="en",
country=region
)
formatted_content = {
"query": query,
"results": [result.model_dump() for result in search_results],
"count": len(search_results)
}
logging.info(f"✅ Found {len(search_results)} results in {search_time:.2f}s")
action_response = ActionResponse(
success=True,
message=formatted_content,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Search operation failed: {str(e)}"
logging.error(f"Search error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "search_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def download_file(
url: str,
output_path: str,
overwrite: bool = False,
timeout: int = 180
) -> Union[str, TextContent]:
"""
Download a file from a URL.
Args:
url: URL to download from
output_path: Local path to save the file
overwrite: Whether to overwrite existing files
timeout: Download timeout in seconds
Returns:
TextContent with download result
"""
try:
if not url.startswith(("http://", "https://")):
raise ValueError("Only HTTP/HTTPS URLs are supported")
output_file = Path(output_path).expanduser().resolve()
if output_file.exists() and not overwrite:
raise ValueError(f"File already exists: {output_file}. Use overwrite=True to replace.")
output_file.parent.mkdir(parents=True, exist_ok=True)
logging.info(f"📥 Downloading from: {url}")
start_time = time.time()
temp_path, content = download_file_from_url(url, timeout=timeout)
# Move to final destination
with open(output_file, 'wb') as f:
f.write(content)
# Clean up temp file
Path(temp_path).unlink(missing_ok=True)
duration = time.time() - start_time
file_size = len(content)
logging.info(f"✅ Downloaded {file_size / 1024:.2f} KB in {duration:.2f}s")
action_response = ActionResponse(
success=True,
message=f"Successfully downloaded file to {output_file}",
metadata={
"url": url,
"output_path": str(output_file),
"file_size_bytes": file_size,
"duration_seconds": duration
}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Download failed: {str(e)}"
logging.error(f"Download error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "download_error", "url": url}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def search_knowledge_base(
query: str,
knowledge_base_path: str,
top_k: int = 5
) -> Union[str, TextContent]:
"""
Search a local knowledge base using simple text matching.
Args:
query: Search query
knowledge_base_path: Path to knowledge base directory
top_k: Number of top results to return
Returns:
TextContent with search results
"""
try:
kb_path = Path(knowledge_base_path).expanduser().resolve()
if not kb_path.exists():
raise FileNotFoundError(f"Knowledge base not found: {kb_path}")
if not kb_path.is_dir():
raise ValueError(f"Knowledge base path must be a directory: {kb_path}")
logging.info(f"🔍 Searching knowledge base: {kb_path}")
# Simple file search - find files containing the query
results = []
query_lower = query.lower()
for file_path in kb_path.rglob("*"):
if file_path.is_file() and file_path.suffix in [".txt", ".md", ".json"]:
try:
content = file_path.read_text(encoding="utf-8", errors="ignore")
if query_lower in content.lower():
# Get snippet around first occurrence
idx = content.lower().index(query_lower)
start = max(0, idx - 100)
end = min(len(content), idx + 200)
snippet = content[start:end].strip()
results.append({
"file": str(file_path.relative_to(kb_path)),
"snippet": snippet,
"relevance": content.lower().count(query_lower)
})
except Exception as e:
logging.warning(f"Error reading {file_path}: {e}")
continue
# Sort by relevance and limit
results.sort(key=lambda x: x["relevance"], reverse=True)
results = results[:top_k]
logging.info(f"✅ Found {len(results)} results")
action_response = ActionResponse(
success=True,
message={
"query": query,
"results": results,
"total_found": len(results)
},
metadata={
"knowledge_base": str(kb_path),
"top_k": top_k
}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Knowledge base search failed: {str(e)}"
logging.error(f"KB search error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "kb_search_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/wayback_enhanced.py¶
"""
Enhanced Wayback Machine tools.
"""
import json
import logging
import traceback
from typing import Union
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from mcp.types import TextContent
from waybackpy import WaybackMachineCDXServerAPI
from base import ActionResponse
load_dotenv()
async def get_archived_content(
url: str,
timestamp: str
) -> Union[str, TextContent]:
"""
Get content from archived webpage.
Args:
url: URL to retrieve
timestamp: Wayback timestamp (YYYYMMDDhhmmss)
Returns:
TextContent with archived content
"""
try:
logging.info(f"🕰️ Getting archived content: {url} at {timestamp}")
# Query for closest snapshot
cdx_api = WaybackMachineCDXServerAPI(url)
snapshot = cdx_api.near(wayback_machine_timestamp=timestamp)
if not snapshot:
raise ValueError("No archived version found")
# Fetch content
response = requests.get(snapshot.archive_url, timeout=30)
response.raise_for_status()
# Extract text
soup = BeautifulSoup(response.content, 'html.parser')
text = soup.get_text(separator=" ", strip=True)
result = {
"url": url,
"timestamp": timestamp,
"actual_timestamp": snapshot.timestamp,
"archive_url": snapshot.archive_url,
"content": text[:10000], # Limit to 10k chars
"content_length": len(text)
}
action_response = ActionResponse(
success=True,
message=result,
metadata={"url": url}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Failed: {str(e)}",
metadata={"error_type": "wayback_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/wiki_enhanced.py¶
"""
Enhanced Wikipedia tools with full article access.
Based on AWorld wiki-server complete implementation.
"""
import json
import logging
import traceback
import calendar
from datetime import datetime
from typing import Union
import requests
import wikipedia
from dotenv import load_dotenv
from mcp.types import TextContent
from base import ActionResponse
load_dotenv()
async def get_article_content(
title: str,
language: str = "en"
) -> Union[str, TextContent]:
"""
Get full Wikipedia article content.
Args:
title: Article title
language: Language code
Returns:
TextContent with full article
"""
try:
wikipedia.set_lang(language)
logging.info(f"📚 Getting full article: {title}")
page = wikipedia.page(title, auto_suggest=True)
result = {
"title": page.title,
"url": page.url,
"content": page.content,
"summary": page.summary,
"categories": page.categories[:20] if page.categories else [],
"links": page.links[:50] if page.links else [],
"images": page.images[:10] if page.images else []
}
logging.info(f"✅ Retrieved article: {len(page.content)} chars")
action_response = ActionResponse(
success=True,
message=result,
metadata={"language": language, "title": page.title}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to get article: {str(e)}"
logging.error(f"Wiki error: {traceback.format_exc()}")
action_response = ActionResponse(
success=False,
message=error_msg,
metadata={"error_type": "wiki_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_article_categories(
title: str,
language: str = "en"
) -> Union[str, TextContent]:
"""
Get categories for Wikipedia article.
Args:
title: Article title
language: Language code
Returns:
TextContent with categories
"""
try:
wikipedia.set_lang(language)
page = wikipedia.page(title, auto_suggest=True)
result = {
"title": page.title,
"categories": page.categories,
"count": len(page.categories)
}
action_response = ActionResponse(
success=True,
message=result,
metadata={"language": language}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Failed: {str(e)}",
metadata={"error_type": "wiki_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_article_links(
title: str,
language: str = "en"
) -> Union[str, TextContent]:
"""
Get links from Wikipedia article.
Args:
title: Article title
language: Language code
Returns:
TextContent with links
"""
try:
wikipedia.set_lang(language)
page = wikipedia.page(title, auto_suggest=True)
result = {
"title": page.title,
"links": page.links[:100], # Limit to 100
"total_links": len(page.links),
"count": min(100, len(page.links))
}
action_response = ActionResponse(
success=True,
message=result,
metadata={"language": language}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Failed: {str(e)}",
metadata={"error_type": "wiki_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_article_history(
title: str,
date: str,
language: str = "en"
) -> Union[str, TextContent]:
"""
Get historical version of Wikipedia article.
Args:
title: Article title
date: Target date (YYYY/MM/DD)
language: Language code
Returns:
TextContent with historical content
"""
try:
logging.info(f"📚 Getting historical version: {title} at {date}")
# Parse date (YYYY/MM or YYYY/MM/DD)
if not isinstance(date, str) or "/" not in date:
raise ValueError("date must be YYYY/MM/DD or YYYY/MM")
date_parts = date.split("/")
if len(date_parts) < 2:
raise ValueError("date must be YYYY/MM/DD or YYYY/MM")
year = int(date_parts[0])
month = int(date_parts[1])
day = int(date_parts[2]) if len(date_parts) > 2 else calendar.monthrange(year, month)[1]
target_date = datetime(year, month, day)
# Get page revisions via Wikipedia API
params = {
"action": "query",
"prop": "revisions",
"titles": title,
"rvprop": "ids|timestamp|user|comment|content",
"rvlimit": 1,
"rvdir": "older",
"rvstart": target_date.isoformat(),
"format": "json"
}
api_url = f"https://{language}.wikipedia.org/w/api.php"
response = requests.get(api_url, params=params, timeout=10)
data = response.json()
page = next(iter(data["query"]["pages"].values()))
if "revisions" in page:
revision = page["revisions"][0]
actual_date = datetime.fromisoformat(revision["timestamp"].replace("Z", "+00:00"))
result = {
"title": title,
"requested_date": date,
"actual_date": actual_date.strftime("%Y/%m/%d"),
"content": revision["*"],
"editor": revision["user"],
"comment": revision.get("comment", "")
}
action_response = ActionResponse(
success=True,
message=result,
metadata={"language": language}
)
else:
action_response = ActionResponse(
success=False,
message="No historical version found",
metadata={"error_type": "not_found"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
action_response = ActionResponse(
success=False,
message=f"Failed: {str(e)}",
metadata={"error_type": "wiki_error"}
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
src/yahoo_finance_tools.py¶
"""
Yahoo Finance comprehensive tools.
Based on AWorld MCP server implementation.
Provides stock quotes, historical data, company info, and financial statements.
"""
import json
import logging
import time
import traceback
from datetime import datetime
from typing import Union, Literal
import yfinance as yf
from dotenv import load_dotenv
from mcp.types import TextContent
from pydantic import BaseModel, Field
from base import ActionResponse
load_dotenv()
class YFinanceMetadata(BaseModel):
"""Metadata for Yahoo Finance operation results."""
symbol: str
operation: str
execution_time: float | None = None
data_points: int | None = None
error_type: str | None = None
timestamp: str | None = None
async def get_stock_quote(
symbol: str
) -> Union[str, TextContent]:
"""
Get current stock quote information.
Args:
symbol: Stock ticker symbol (e.g., AAPL, MSFT)
Returns:
TextContent with quote data
"""
try:
start_time = time.time()
logging.info(f"📈 Fetching stock quote for: {symbol}")
ticker = yf.Ticker(symbol)
info = ticker.info
if not info or (info.get("regularMarketPrice") is None and info.get("currentPrice") is None):
# Try to get basic history to validate symbol
hist = ticker.history(period="1d")
if hist.empty:
raise ValueError(f"No data found for symbol: {symbol}")
raise ValueError(f"Could not retrieve detailed quote for symbol: {symbol}")
# Extract key quote information
quote_data = {
"symbol": symbol.upper(),
"company_name": info.get("shortName", info.get("longName")),
"current_price": info.get("regularMarketPrice", info.get("currentPrice")),
"previous_close": info.get("previousClose"),
"open": info.get("regularMarketOpen", info.get("open")),
"day_high": info.get("regularMarketDayHigh", info.get("dayHigh")),
"day_low": info.get("regularMarketDayLow", info.get("dayLow")),
"volume": info.get("regularMarketVolume", info.get("volume")),
"average_volume": info.get("averageVolume"),
"market_cap": info.get("marketCap"),
"fifty_two_week_high": info.get("fiftyTwoWeekHigh"),
"fifty_two_week_low": info.get("fiftyTwoWeekLow"),
"currency": info.get("currency"),
"exchange": info.get("exchange")
}
# Filter out None values
quote_data = {k: v for k, v in quote_data.items() if v is not None}
# Calculate change
if quote_data.get("current_price") and quote_data.get("previous_close"):
change = quote_data["current_price"] - quote_data["previous_close"]
change_pct = (change / quote_data["previous_close"]) * 100
quote_data["change"] = round(change, 2)
quote_data["change_percent"] = round(change_pct, 2)
execution_time = time.time() - start_time
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_stock_quote",
execution_time=execution_time,
data_points=len(quote_data),
timestamp=datetime.now().isoformat()
)
logging.info(f"✅ Stock quote: ${quote_data.get('current_price')}")
action_response = ActionResponse(
success=True,
message=quote_data,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to fetch stock quote: {str(e)}"
logging.error(f"Stock quote error: {traceback.format_exc()}")
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_stock_quote",
error_type=type(e).__name__,
timestamp=datetime.now().isoformat()
)
action_response = ActionResponse(
success=False,
message=error_msg,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_historical_data(
symbol: str,
start: str,
end: str,
interval: str = "1d",
max_rows_preview: int = 10
) -> Union[str, TextContent]:
"""
Retrieve historical stock data.
Args:
symbol: Stock ticker symbol
start: Start date (YYYY-MM-DD)
end: End date (YYYY-MM-DD)
interval: Data interval (1d, 1wk, 1mo, etc.)
max_rows_preview: Maximum rows to show in preview (0 for all)
Returns:
TextContent with historical data
"""
try:
start_time = time.time()
logging.info(f"📈 Fetching historical data for: {symbol}")
ticker = yf.Ticker(symbol)
hist_df = ticker.history(start=start, end=end, interval=interval)
if hist_df.empty:
raise ValueError(f"No historical data found for {symbol}")
# Convert DataFrame to list of dictionaries
hist_df.reset_index(inplace=True)
# Ensure date columns are strings
if "Date" in hist_df.columns:
hist_df["Date"] = hist_df["Date"].astype(str)
if "Datetime" in hist_df.columns:
hist_df["Datetime"] = hist_df["Datetime"].astype(str)
# Clean column names
hist_df.columns = hist_df.columns.str.replace(" ", "")
historical_data = hist_df.to_dict(orient="records")
execution_time = time.time() - start_time
# Prepare result with preview
result = {
"symbol": symbol.upper(),
"start_date": start,
"end_date": end,
"interval": interval,
"total_records": len(historical_data),
"data": historical_data if max_rows_preview == 0 else historical_data[:max_rows_preview]
}
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_historical_data",
execution_time=execution_time,
data_points=len(historical_data),
timestamp=datetime.now().isoformat()
)
logging.info(f"✅ Retrieved {len(historical_data)} historical records")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to fetch historical data: {str(e)}"
logging.error(f"Historical data error: {traceback.format_exc()}")
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_historical_data",
error_type=type(e).__name__,
timestamp=datetime.now().isoformat()
)
action_response = ActionResponse(
success=False,
message=error_msg,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_company_info(
symbol: str
) -> Union[str, TextContent]:
"""
Get company information and business details.
Args:
symbol: Stock ticker symbol
Returns:
TextContent with company information
"""
try:
start_time = time.time()
logging.info(f"🏢 Fetching company info for: {symbol}")
ticker = yf.Ticker(symbol)
info = ticker.info
if not info or not info.get("symbol"):
raise ValueError(f"No company information found for symbol: {symbol}")
# Extract key company information
company_data = {
"symbol": info.get("symbol"),
"short_name": info.get("shortName"),
"long_name": info.get("longName"),
"sector": info.get("sector"),
"industry": info.get("industry"),
"full_time_employees": info.get("fullTimeEmployees"),
"business_summary": info.get("longBusinessSummary"),
"city": info.get("city"),
"state": info.get("state"),
"country": info.get("country"),
"website": info.get("website"),
"exchange": info.get("exchange"),
"currency": info.get("currency"),
"market_cap": info.get("marketCap"),
"pe_ratio": info.get("trailingPE"),
"forward_pe": info.get("forwardPE"),
"dividend_yield": info.get("dividendYield"),
"beta": info.get("beta")
}
# Filter out None values
company_data = {k: v for k, v in company_data.items() if v is not None}
execution_time = time.time() - start_time
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_company_info",
execution_time=execution_time,
data_points=len(company_data),
timestamp=datetime.now().isoformat()
)
logging.info(f"✅ Retrieved company info: {company_data.get('long_name', company_data.get('short_name'))}")
action_response = ActionResponse(
success=True,
message=company_data,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to fetch company info: {str(e)}"
logging.error(f"Company info error: {traceback.format_exc()}")
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_company_info",
error_type=type(e).__name__,
timestamp=datetime.now().isoformat()
)
action_response = ActionResponse(
success=False,
message=error_msg,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
async def get_financial_statements(
symbol: str,
statement_type: Literal["income_statement", "balance_sheet", "cash_flow"],
period_type: Literal["annual", "quarterly"] = "annual",
max_columns_preview: int = 4
) -> Union[str, TextContent]:
"""
Get financial statements for a company.
Args:
symbol: Stock ticker symbol
statement_type: Type of statement (income_statement, balance_sheet, cash_flow)
period_type: Period type (annual or quarterly)
max_columns_preview: Maximum periods to show (0 for all)
Returns:
TextContent with financial statement data
"""
try:
start_time = time.time()
logging.info(f"📋 Fetching {statement_type} for: {symbol}")
ticker = yf.Ticker(symbol)
statement_df = None
# Get appropriate statement
if statement_type == "income_statement":
statement_df = ticker.income_stmt if period_type == "annual" else ticker.quarterly_income_stmt
elif statement_type == "balance_sheet":
statement_df = ticker.balance_sheet if period_type == "annual" else ticker.quarterly_balance_sheet
elif statement_type == "cash_flow":
statement_df = ticker.cashflow if period_type == "annual" else ticker.quarterly_cashflow
else:
raise ValueError(f"Invalid statement_type: {statement_type}")
if statement_df is None or statement_df.empty:
raise ValueError(f"No {period_type} {statement_type} data found for symbol {symbol}")
# Process DataFrame
statement_df.reset_index(inplace=True)
statement_df.rename(columns={"index": "Item"}, inplace=True)
# Convert date columns to strings
for col in statement_df.columns:
if col != "Item":
try:
if hasattr(col, "strftime"):
new_col_name = col.strftime("%Y-%m-%d")
statement_df.rename(columns={col: new_col_name}, inplace=True)
except Exception:
pass
# Limit columns if needed
if max_columns_preview > 0 and len(statement_df.columns) > (max_columns_preview + 1):
columns_to_keep = ["Item"] + list(statement_df.columns[1:max_columns_preview + 1])
statement_df = statement_df[columns_to_keep]
statement_data = statement_df.to_dict(orient="records")
execution_time = time.time() - start_time
result = {
"symbol": symbol.upper(),
"statement_type": statement_type,
"period_type": period_type,
"total_line_items": len(statement_data),
"periods": len(statement_df.columns) - 1,
"data": statement_data
}
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_financial_statements",
execution_time=execution_time,
data_points=len(statement_data),
timestamp=datetime.now().isoformat()
)
logging.info(f"✅ Retrieved {statement_type}: {len(statement_data)} items")
action_response = ActionResponse(
success=True,
message=result,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
except Exception as e:
error_msg = f"Failed to fetch financial statements: {str(e)}"
logging.error(f"Financial statements error: {traceback.format_exc()}")
metadata = YFinanceMetadata(
symbol=symbol.upper(),
operation="get_financial_statements",
error_type=type(e).__name__,
timestamp=datetime.now().isoformat()
)
action_response = ActionResponse(
success=False,
message=error_msg,
metadata=metadata.model_dump()
)
return TextContent(
type="text",
text=json.dumps(action_response.model_dump())
)
test_document_tools.py¶
"""
Tests for document processing tools.
Uses real files for testing.
"""
import asyncio
import json
import pytest
import tempfile
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent / "src"))
from document_processing_tools import (
extract_pdf_text,
extract_docx_content,
extract_pptx_content,
extract_csv_content
)
class TestPDFExtraction:
"""Tests for PDF extraction."""
@pytest.mark.asyncio
async def test_extract_pdf_basic(self):
"""Test basic PDF extraction."""
# Create a simple PDF for testing
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
pdf_path = Path(tempfile.mktemp(suffix=".pdf"))
# Create PDF
c = canvas.Canvas(str(pdf_path), pagesize=letter)
c.drawString(100, 750, "Hello World")
c.drawString(100, 730, "This is a test PDF")
c.showPage()
c.drawString(100, 750, "Page 2 content")
c.save()
try:
result = await extract_pdf_text(str(pdf_path))
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["file_type"] == "pdf"
assert message["total_pages"] == 2
assert "Hello World" in message["text"]
print(f"✅ PDF extraction successful: {message['total_pages']} pages")
finally:
pdf_path.unlink(missing_ok=True)
@pytest.mark.asyncio
async def test_extract_pdf_page_range(self):
"""Test PDF extraction with page range."""
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
pdf_path = Path(tempfile.mktemp(suffix=".pdf"))
c = canvas.Canvas(str(pdf_path), pagesize=letter)
for i in range(1, 6):
c.drawString(100, 750, f"Page {i}")
c.showPage()
c.save()
try:
result = await extract_pdf_text(str(pdf_path), page_range="1-3")
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["pages_extracted"] == 3
print(f"✅ PDF page range: extracted {message['pages_extracted']} pages")
finally:
pdf_path.unlink(missing_ok=True)
class TestDOCXExtraction:
"""Tests for DOCX extraction."""
@pytest.mark.asyncio
async def test_extract_docx(self):
"""Test DOCX extraction."""
from docx import Document
docx_path = Path(tempfile.mktemp(suffix=".docx"))
# Create DOCX
doc = Document()
doc.add_paragraph("First paragraph")
doc.add_paragraph("Second paragraph")
doc.save(docx_path)
try:
result = await extract_docx_content(str(docx_path))
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["file_type"] == "docx"
assert message["paragraphs"] >= 2
assert "First paragraph" in message["text"]
print(f"✅ DOCX extraction: {message['paragraphs']} paragraphs")
finally:
docx_path.unlink(missing_ok=True)
class TestPPTXExtraction:
"""Tests for PPTX extraction."""
@pytest.mark.asyncio
async def test_extract_pptx(self):
"""Test PPTX extraction."""
from pptx import Presentation
pptx_path = Path(tempfile.mktemp(suffix=".pptx"))
# Create PPTX
prs = Presentation()
slide = prs.slides.add_slide(prs.slide_layouts[1])
slide.shapes.title.text = "Test Slide"
prs.save(pptx_path)
try:
result = await extract_pptx_content(str(pptx_path))
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["file_type"] == "pptx"
assert message["total_slides"] >= 1
print(f"✅ PPTX extraction: {message['total_slides']} slides")
finally:
pptx_path.unlink(missing_ok=True)
class TestCSVExtraction:
"""Tests for CSV extraction."""
@pytest.mark.asyncio
async def test_extract_csv(self):
"""Test CSV extraction."""
csv_path = Path(tempfile.mktemp(suffix=".csv"))
# Create CSV
csv_content = """name,age,city
Alice,30,New York
Bob,25,Los Angeles
Charlie,35,Chicago"""
csv_path.write_text(csv_content)
try:
result = await extract_csv_content(str(csv_path))
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["file_type"] == "csv"
assert message["rows"] == 3
assert message["columns"] == 3
assert "name" in message["column_names"]
print(f"✅ CSV extraction: {message['rows']} rows, {message['columns']} columns")
finally:
csv_path.unlink(missing_ok=True)
if __name__ == "__main__":
print("=" * 70)
print("Running Document Processing Tools Tests")
print("=" * 70)
print()
pytest.main([__file__, "-v", "-s"])
test_imports.py¶
"""
Test script to verify all imports and basic module loading.
"""
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
print("Testing imports...")
try:
print("\n✓ Importing base module...")
from base import ActionResponse, DocumentMetadata, is_url, validate_file_path
print("✓ Importing search_tools module...")
from search_tools import search_web, download_file, search_knowledge_base
print("✓ Importing multimodal_tools module...")
from multimodal_tools import read_webpage, read_document, parse_image, parse_video
print("✓ Importing filesystem_tools module...")
from filesystem_tools import read_file, grep_search, summarize_text
print("✓ Importing public_data_tools module...")
from public_data_tools import (
get_weather, get_stock_price, convert_currency,
search_wikipedia, search_arxiv, search_wayback
)
print("✓ Importing private_data_tools module...")
from private_data_tools import get_calendar_events, search_notion
print("✓ Importing main module...")
from main import mcp
print("\n" + "="*80)
print("✅ All imports successful!")
print("="*80)
# Test basic functionality
print("\nTesting basic functionality...")
# Test ActionResponse
response = ActionResponse(
success=True,
message="Test message",
metadata={"test": "value"}
)
assert response.success == True
print("✓ ActionResponse working")
# Test is_url
assert is_url("https://example.com") == True
assert is_url("/path/to/file") == False
print("✓ is_url working")
print("\n" + "="*80)
print("✅ All tests passed!")
print("="*80)
print("\nℹ️ The MCP server is ready to use.")
print(" Run 'python src/main.py' to start the server.")
print(" Run 'python quickstart.py' to test various tools.")
except ImportError as e:
print(f"\n❌ Import error: {e}")
print("\nℹ️ You may need to install dependencies:")
print(" pip install -r requirements.txt")
sys.exit(1)
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
test_new_tools.py¶
"""
Test script for new perception tools: crypto prices, location search, and POI search.
"""
import asyncio
import json
import logging
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from public_data_tools import get_crypto_price, search_location, search_poi
logging.basicConfig(level=logging.INFO)
async def test_new_tools():
"""Test the new perception tools."""
print("\n" + "="*80)
print("NEW PERCEPTION TOOLS - TEST")
print("="*80)
# Test 1: Crypto Price
print("\n📝 Test 1: Cryptocurrency Price (Bitcoin)")
print("-" * 80)
try:
result = await get_crypto_price("btc", "usd")
data = json.loads(result.text)
if data['success']:
msg = data['message']
print(f"✅ {msg['symbol']}: ${msg['current_price']:,.2f} USD")
print(f" 24h Change: {msg['price_change_24h_percent']:.2f}%")
print(f" Market Cap: ${msg['market_cap']:,.0f}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 2: Crypto Price - Ethereum
print("\n📝 Test 2: Cryptocurrency Price (Ethereum)")
print("-" * 80)
try:
result = await get_crypto_price("eth", "eur")
data = json.loads(result.text)
if data['success']:
msg = data['message']
print(f"✅ {msg['symbol']}: €{msg['current_price']:,.2f} EUR")
print(f" 24h Volume: €{msg['volume_24h']:,.0f}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 3: Location Search
print("\n📝 Test 3: Location Search (Eiffel Tower)")
print("-" * 80)
try:
result = await search_location("Eiffel Tower", limit=3)
data = json.loads(result.text)
if data['success']:
locations = data['message']['locations']
print(f"✅ Found {len(locations)} locations:")
for loc in locations[:2]:
print(f" - {loc['display_name']}")
print(f" Coordinates: ({loc['latitude']}, {loc['longitude']})")
except Exception as e:
print(f"❌ Error: {e}")
# Test 4: Location Search with Country Filter
print("\n📝 Test 4: Location Search (Paris, France only)")
print("-" * 80)
try:
result = await search_location("Paris", limit=3, country_code="fr")
data = json.loads(result.text)
if data['success']:
locations = data['message']['locations']
print(f"✅ Found {len(locations)} locations in France:")
for loc in locations[:2]:
print(f" - {loc['display_name']}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 5: POI Search (Restaurants near Eiffel Tower)
print("\n📝 Test 5: POI Search (Restaurants near Eiffel Tower)")
print("-" * 80)
try:
# Eiffel Tower coordinates
latitude = 48.8584
longitude = 2.2945
result = await search_poi("restaurant", latitude, longitude, radius=500, limit=5)
data = json.loads(result.text)
if data['success']:
pois = data['message']['pois']
print(f"✅ Found {len(pois)} restaurants:")
for poi in pois[:3]:
print(f" - {poi['name']}")
if poi.get('cuisine'):
print(f" Cuisine: {poi['cuisine']}")
except Exception as e:
print(f"❌ Error: {e}")
# Test 6: POI Search (Coffee shops in San Francisco)
print("\n📝 Test 6: POI Search (Coffee shops in San Francisco)")
print("-" * 80)
try:
# San Francisco downtown coordinates
latitude = 37.7749
longitude = -122.4194
result = await search_poi("cafe", latitude, longitude, radius=1000, limit=5)
data = json.loads(result.text)
if data['success']:
pois = data['message']['pois']
print(f"✅ Found {len(pois)} cafes:")
for poi in pois[:3]:
print(f" - {poi['name']}")
if poi.get('opening_hours'):
print(f" Hours: {poi['opening_hours']}")
except Exception as e:
print(f"❌ Error: {e}")
print("\n" + "="*80)
print("TEST COMPLETE")
print("="*80)
print("\nℹ️ All tests use free, open APIs - no API keys required!")
print("\n")
if __name__ == "__main__":
asyncio.run(test_new_tools())
test_pubchem_tools.py¶
"""
Real API tests for PubChem tools.
These tests make actual API calls to PubChem to verify functionality.
"""
import asyncio
import json
import pytest
from pathlib import Path
import sys
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from pubchem_tools import (
search_compounds,
get_compound_properties,
get_compound_synonyms,
search_similar_compounds
)
class TestPubChemSearch:
"""Tests for compound search functionality."""
@pytest.mark.asyncio
async def test_search_by_name(self):
"""Test searching compounds by name."""
result = await search_compounds(
query="aspirin",
search_type="name",
max_results=5
)
# Parse result
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["query"] == "aspirin"
assert message["search_type"] == "name"
assert len(message["compounds"]) > 0
# Check first compound has expected fields
compound = message["compounds"][0]
assert compound["cid"] is not None
assert compound["name"] is not None
assert compound["molecular_formula"] is not None
assert compound["molecular_weight"] is not None
print(f"✅ Found {len(message['compounds'])} compounds for 'aspirin'")
print(f" First: {compound['name']} (CID: {compound['cid']})")
@pytest.mark.asyncio
async def test_search_by_cid(self):
"""Test searching compound by CID."""
result = await search_compounds(
query="2244", # Aspirin CID
search_type="cid",
max_results=1
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert len(message["compounds"]) == 1
compound = message["compounds"][0]
assert compound["cid"] == 2244
assert "aspirin" in compound["name"].lower() or "acetylsalicylic" in compound["name"].lower()
print(f"✅ Found compound by CID: {compound['name']}")
@pytest.mark.asyncio
async def test_search_by_formula(self):
"""Test searching compounds by molecular formula."""
result = await search_compounds(
query="C9H8O4", # Aspirin formula
search_type="formula",
max_results=10
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert len(message["compounds"]) > 0
# Check all compounds have the correct formula
for compound in message["compounds"]:
assert compound["molecular_formula"] == "C9H8O4"
print(f"✅ Found {len(message['compounds'])} compounds with formula C9H8O4")
@pytest.mark.asyncio
async def test_search_by_smiles(self):
"""Test searching compound by SMILES."""
result = await search_compounds(
query="CC(=O)OC1=CC=CC=C1C(=O)O", # Aspirin SMILES
search_type="smiles",
max_results=1
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert len(message["compounds"]) > 0
print(f"✅ Found compound by SMILES")
@pytest.mark.asyncio
async def test_search_invalid_query(self):
"""Test searching with invalid query."""
result = await search_compounds(
query="",
search_type="name",
max_results=5
)
data = json.loads(result.text)
assert data["success"] is False
assert "required" in data["message"].lower()
print("✅ Correctly handled invalid query")
@pytest.mark.asyncio
async def test_search_not_found(self):
"""Test searching for non-existent compound."""
result = await search_compounds(
query="xyzabc123notarealcompound999",
search_type="name",
max_results=5
)
data = json.loads(result.text)
# Should fail or return empty results
if data["success"]:
assert data["message"]["count"] == 0
print("✅ Handled non-existent compound search")
class TestPubChemProperties:
"""Tests for compound properties functionality."""
@pytest.mark.asyncio
async def test_get_properties(self):
"""Test getting compound properties."""
result = await get_compound_properties(
cid=2244, # Aspirin
properties=["MolecularWeight", "MolecularFormula", "XLogP", "TPSA"]
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["cid"] == 2244
props = message["properties"]
assert "MolecularWeight" in props
assert "MolecularFormula" in props
assert props["MolecularFormula"] == "C9H8O4"
print(f"✅ Retrieved properties for aspirin:")
print(f" Formula: {props['MolecularFormula']}")
print(f" Weight: {props['MolecularWeight']}")
@pytest.mark.asyncio
async def test_get_default_properties(self):
"""Test getting default properties."""
result = await get_compound_properties(
cid=2244 # Aspirin
)
data = json.loads(result.text)
assert data["success"] is True
props = data["message"]["properties"]
# Should have default properties
assert len(props) > 0
assert "MolecularWeight" in props
print(f"✅ Retrieved {len(props)} default properties")
@pytest.mark.asyncio
async def test_get_properties_invalid_cid(self):
"""Test getting properties with invalid CID."""
result = await get_compound_properties(
cid=-1
)
data = json.loads(result.text)
assert data["success"] is False
print("✅ Correctly handled invalid CID")
class TestPubChemSynonyms:
"""Tests for compound synonyms functionality."""
@pytest.mark.asyncio
async def test_get_synonyms(self):
"""Test getting compound synonyms."""
result = await get_compound_synonyms(
cid=2244, # Aspirin
max_synonyms=10
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["cid"] == 2244
assert len(message["synonyms"]) > 0
# Check that common names are included
synonyms_lower = [s.lower() for s in message["synonyms"]]
assert any("aspirin" in s for s in synonyms_lower)
print(f"✅ Retrieved {len(message['synonyms'])} synonyms:")
print(f" Examples: {', '.join(message['synonyms'][:3])}")
@pytest.mark.asyncio
async def test_get_synonyms_limit(self):
"""Test synonym count limit."""
result = await get_compound_synonyms(
cid=2244,
max_synonyms=5
)
data = json.loads(result.text)
assert data["success"] is True
synonyms = data["message"]["synonyms"]
assert len(synonyms) <= 5
print(f"✅ Correctly limited synonyms to {len(synonyms)}")
class TestPubChemSimilarity:
"""Tests for similar compounds search."""
@pytest.mark.asyncio
async def test_search_similar(self):
"""Test searching similar compounds."""
result = await search_similar_compounds(
cid=2244, # Aspirin
similarity_threshold=0.9,
max_results=5
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["reference_cid"] == 2244
assert message["similarity_threshold"] == 0.9
similar = message["similar_compounds"]
# Should find at least some similar compounds
if len(similar) > 0:
compound = similar[0]
assert compound["cid"] is not None
assert compound["cid"] != 2244 # Should not include itself
assert compound["name"] is not None
print(f"✅ Found {len(similar)} similar compounds:")
print(f" Example: {compound['name']} (CID: {compound['cid']})")
else:
print("✅ Search completed (no similar compounds at 0.9 threshold)")
@pytest.mark.asyncio
async def test_search_similar_lower_threshold(self):
"""Test searching similar compounds with lower threshold."""
result = await search_similar_compounds(
cid=2244,
similarity_threshold=0.7, # Lower threshold
max_results=10
)
data = json.loads(result.text)
assert data["success"] is True
similar = data["message"]["similar_compounds"]
# With lower threshold, should find more compounds
print(f"✅ Found {len(similar)} compounds at 0.7 similarity")
class TestPubChemRateLimit:
"""Tests for rate limiting functionality."""
@pytest.mark.asyncio
async def test_multiple_requests(self):
"""Test that multiple requests are rate-limited."""
import time
start_time = time.time()
# Make 5 requests in quick succession
for i in range(5):
result = await search_compounds(
query="aspirin",
search_type="name",
max_results=1
)
data = json.loads(result.text)
assert data["success"] is True
elapsed = time.time() - start_time
# Should take at least 0.8 seconds (5 requests * 0.2s delay - 0.2s for first)
assert elapsed >= 0.8
print(f"✅ Rate limiting working: {elapsed:.2f}s for 5 requests")
# Run tests
if __name__ == "__main__":
print("=" * 70)
print("Running PubChem Tools Real API Tests")
print("=" * 70)
print()
# Run with pytest
pytest.main([__file__, "-v", "-s"])
test_video_keyframes_num_frames.py¶
"""Regression test: num_frames=0 must not cause ZeroDivisionError.
The LLM-supplied num_frames parameter was used directly as a divisor in
`frame_count // num_frames`; num_frames=0 crashed with ZeroDivisionError
(surfacing as a confusing tool error). It is now clamped to >= 1 up front.
"""
import asyncio
import json
import os
import sys
import types
from types import SimpleNamespace
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src"))
# Optional runtime deps for importing the chapter module in unit tests.
sys.modules.setdefault("dotenv", types.SimpleNamespace(load_dotenv=lambda: None))
mcp = types.ModuleType("mcp")
mcp_types = types.ModuleType("mcp.types")
class TextContent:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
mcp_types.TextContent = TextContent
sys.modules["mcp"] = mcp
sys.modules["mcp.types"] = mcp_types
import cv2
import numpy as np
import media_processing_tools
from media_processing_tools import extract_video_keyframes
def _make_clip(path, frames=20):
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
out = cv2.VideoWriter(str(path), fourcc, 10.0, (64, 48))
for _ in range(frames):
out.write(np.zeros((48, 64, 3), dtype=np.uint8))
out.release()
def test_extract_keyframes_zero_num_frames_is_clamped(tmp_path):
clip = tmp_path / "clip.mp4"
_make_clip(clip)
result = asyncio.run(extract_video_keyframes(str(clip), num_frames=0))
payload = json.loads(result.text)
assert payload["success"] is True
assert "division" not in str(payload["message"]).lower()
def test_analyze_video_ai_zero_num_frames_is_clamped(tmp_path, monkeypatch):
clip = tmp_path / "clip.mp4"
_make_clip(clip)
message = SimpleNamespace(content="a frame")
response = SimpleNamespace(choices=[SimpleNamespace(message=message)])
client = SimpleNamespace(chat=SimpleNamespace(
completions=SimpleNamespace(create=lambda **kwargs: response)))
monkeypatch.setattr(media_processing_tools, "_make_vision_client",
lambda: (client, "fake-model"))
result = asyncio.run(media_processing_tools.analyze_video_ai(str(clip), num_frames=0))
payload = json.loads(result.text)
assert payload["success"] is True
assert "division" not in str(payload["message"]).lower()
test_wiki_article_history_date.py¶
import asyncio
import json
import os
import sys
import types
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src"))
# Optional runtime deps for importing the chapter module in unit tests.
sys.modules.setdefault("wikipedia", types.ModuleType("wikipedia"))
sys.modules.setdefault("dotenv", types.SimpleNamespace(load_dotenv=lambda: None))
mcp = types.ModuleType("mcp")
mcp_types = types.ModuleType("mcp.types")
class TextContent:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
mcp_types.TextContent = TextContent
sys.modules["mcp"] = mcp
sys.modules["mcp.types"] = mcp_types
from wiki_enhanced import get_article_history
def test_year_only_date_returns_error_payload():
result = asyncio.run(get_article_history("Python", "2025"))
payload = json.loads(result.text)
assert payload["success"] is False
msg = str(payload["message"])
assert "date must be" in msg or "Failed" in msg
test_yahoo_finance_tools.py¶
"""
Real API tests for Yahoo Finance tools.
These tests make actual API calls to Yahoo Finance to verify functionality.
"""
import asyncio
import json
import pytest
from pathlib import Path
import sys
from datetime import datetime, timedelta
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from yahoo_finance_tools import (
get_stock_quote,
get_historical_data,
get_company_info,
get_financial_statements
)
class TestYFinanceQuote:
"""Tests for stock quote functionality."""
@pytest.mark.asyncio
async def test_get_stock_quote_aapl(self):
"""Test getting stock quote for AAPL."""
result = await get_stock_quote(symbol="AAPL")
data = json.loads(result.text)
assert data["success"] is True
quote = data["message"]
assert quote["symbol"] == "AAPL"
assert quote["current_price"] is not None
assert quote["current_price"] > 0
assert quote["company_name"] is not None
print(f"✅ AAPL Quote: ${quote['current_price']}")
print(f" Company: {quote['company_name']}")
if "change_percent" in quote:
print(f" Change: {quote['change_percent']}%")
@pytest.mark.asyncio
async def test_get_stock_quote_multiple(self):
"""Test getting quotes for multiple symbols."""
symbols = ["MSFT", "GOOGL", "TSLA"]
for symbol in symbols:
result = await get_stock_quote(symbol=symbol)
data = json.loads(result.text)
assert data["success"] is True
quote = data["message"]
assert quote["symbol"] == symbol
assert quote["current_price"] > 0
print(f"✅ {symbol}: ${quote['current_price']}")
@pytest.mark.asyncio
async def test_get_stock_quote_invalid(self):
"""Test getting quote for invalid symbol."""
result = await get_stock_quote(symbol="INVALIDXYZ999")
data = json.loads(result.text)
assert data["success"] is False
assert "error" in data["message"].lower() or "not found" in data["message"].lower() or "no data" in data["message"].lower()
print("✅ Correctly handled invalid symbol")
@pytest.mark.asyncio
async def test_get_stock_quote_with_metadata(self):
"""Test that quote includes proper metadata."""
result = await get_stock_quote(symbol="AAPL")
data = json.loads(result.text)
assert data["success"] is True
metadata = data["metadata"]
assert metadata["symbol"] == "AAPL"
assert metadata["operation"] == "get_stock_quote"
assert metadata["execution_time"] is not None
assert metadata["execution_time"] > 0
assert metadata["data_points"] > 0
print(f"✅ Metadata OK: {metadata['execution_time']:.2f}s, {metadata['data_points']} fields")
class TestYFinanceHistorical:
"""Tests for historical data functionality."""
@pytest.mark.asyncio
async def test_get_historical_data_1week(self):
"""Test getting 1 week of historical data."""
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
result = await get_historical_data(
symbol="AAPL",
start=start_date.strftime("%Y-%m-%d"),
end=end_date.strftime("%Y-%m-%d"),
interval="1d",
max_rows_preview=10
)
data = json.loads(result.text)
assert data["success"] is True
hist = data["message"]
assert hist["symbol"] == "AAPL"
assert hist["total_records"] > 0
assert len(hist["data"]) > 0
# Check data structure
first_record = hist["data"][0]
assert "Close" in first_record or "close" in str(first_record).lower()
assert "Volume" in first_record or "volume" in str(first_record).lower()
print(f"✅ Retrieved {hist['total_records']} historical records")
print(f" Date range: {hist['start_date']} to {hist['end_date']}")
@pytest.mark.asyncio
async def test_get_historical_data_1month(self):
"""Test getting 1 month of historical data."""
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
result = await get_historical_data(
symbol="MSFT",
start=start_date.strftime("%Y-%m-%d"),
end=end_date.strftime("%Y-%m-%d"),
interval="1d",
max_rows_preview=5
)
data = json.loads(result.text)
assert data["success"] is True
hist = data["message"]
assert hist["total_records"] >= 20 # At least ~20 trading days in a month
print(f"✅ Retrieved {hist['total_records']} records for 1 month period")
@pytest.mark.asyncio
async def test_get_historical_data_weekly(self):
"""Test getting weekly interval data."""
end_date = datetime.now()
start_date = end_date - timedelta(days=90)
result = await get_historical_data(
symbol="AAPL",
start=start_date.strftime("%Y-%m-%d"),
end=end_date.strftime("%Y-%m-%d"),
interval="1wk",
max_rows_preview=10
)
data = json.loads(result.text)
assert data["success"] is True
hist = data["message"]
assert hist["interval"] == "1wk"
print(f"✅ Retrieved {hist['total_records']} weekly records")
class TestYFinanceCompanyInfo:
"""Tests for company information functionality."""
@pytest.mark.asyncio
async def test_get_company_info_aapl(self):
"""Test getting company info for Apple."""
result = await get_company_info(symbol="AAPL")
data = json.loads(result.text)
assert data["success"] is True
info = data["message"]
assert info["symbol"] == "AAPL"
assert info["sector"] is not None
assert info["industry"] is not None
assert info["business_summary"] is not None
assert "apple" in info["business_summary"].lower()
print(f"✅ Company Info for {info.get('long_name', info.get('short_name'))}")
print(f" Sector: {info['sector']}")
print(f" Industry: {info['industry']}")
if "full_time_employees" in info:
print(f" Employees: {info['full_time_employees']:,}")
@pytest.mark.asyncio
async def test_get_company_info_multiple(self):
"""Test getting company info for multiple companies."""
symbols = ["MSFT", "GOOGL", "AMZN"]
for symbol in symbols:
result = await get_company_info(symbol=symbol)
data = json.loads(result.text)
assert data["success"] is True
info = data["message"]
assert info["symbol"] == symbol
assert info["sector"] is not None
print(f"✅ {symbol}: {info.get('long_name', info.get('short_name'))} - {info['sector']}")
@pytest.mark.asyncio
async def test_get_company_info_with_website(self):
"""Test that company info includes website."""
result = await get_company_info(symbol="AAPL")
data = json.loads(result.text)
assert data["success"] is True
info = data["message"]
assert "website" in info
assert "apple.com" in info["website"].lower()
print(f"✅ Website: {info['website']}")
class TestYFinanceFinancialStatements:
"""Tests for financial statements functionality."""
@pytest.mark.asyncio
async def test_get_income_statement(self):
"""Test getting income statement."""
result = await get_financial_statements(
symbol="AAPL",
statement_type="income_statement",
period_type="annual",
max_columns_preview=2
)
data = json.loads(result.text)
assert data["success"] is True
stmt = data["message"]
assert stmt["symbol"] == "AAPL"
assert stmt["statement_type"] == "income_statement"
assert stmt["period_type"] == "annual"
assert len(stmt["data"]) > 0
# Check for key income statement items
items = [item["Item"] for item in stmt["data"]]
# Usually includes items like "Total Revenue", "Net Income", etc.
assert len(items) > 10
print(f"✅ Income Statement: {stmt['total_line_items']} items, {stmt['periods']} periods")
print(f" Sample items: {', '.join(items[:3])}")
@pytest.mark.asyncio
async def test_get_balance_sheet(self):
"""Test getting balance sheet."""
result = await get_financial_statements(
symbol="MSFT",
statement_type="balance_sheet",
period_type="annual",
max_columns_preview=2
)
data = json.loads(result.text)
assert data["success"] is True
stmt = data["message"]
assert stmt["statement_type"] == "balance_sheet"
assert len(stmt["data"]) > 0
print(f"✅ Balance Sheet: {stmt['total_line_items']} items")
@pytest.mark.asyncio
async def test_get_cash_flow(self):
"""Test getting cash flow statement."""
result = await get_financial_statements(
symbol="GOOGL",
statement_type="cash_flow",
period_type="annual",
max_columns_preview=2
)
data = json.loads(result.text)
assert data["success"] is True
stmt = data["message"]
assert stmt["statement_type"] == "cash_flow"
assert len(stmt["data"]) > 0
print(f"✅ Cash Flow: {stmt['total_line_items']} items")
@pytest.mark.asyncio
async def test_get_quarterly_income_statement(self):
"""Test getting quarterly income statement."""
result = await get_financial_statements(
symbol="AAPL",
statement_type="income_statement",
period_type="quarterly",
max_columns_preview=4
)
data = json.loads(result.text)
assert data["success"] is True
stmt = data["message"]
assert stmt["period_type"] == "quarterly"
assert stmt["periods"] >= 4 # Should have at least 4 quarters
print(f"✅ Quarterly Income Statement: {stmt['periods']} quarters")
@pytest.mark.asyncio
async def test_financial_statement_invalid_type(self):
"""Test getting financial statement with invalid type."""
result = await get_financial_statements(
symbol="AAPL",
statement_type="invalid_type", # type: ignore
period_type="annual"
)
data = json.loads(result.text)
assert data["success"] is False
print("✅ Correctly handled invalid statement type")
class TestYFinanceIntegration:
"""Integration tests combining multiple operations."""
@pytest.mark.asyncio
async def test_complete_stock_analysis(self):
"""Test getting complete stock analysis data."""
symbol = "AAPL"
# Get quote
quote_result = await get_stock_quote(symbol)
quote_data = json.loads(quote_result.text)
assert quote_data["success"] is True
# Get company info
info_result = await get_company_info(symbol)
info_data = json.loads(info_result.text)
assert info_data["success"] is True
# Get historical data
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
hist_result = await get_historical_data(
symbol,
start_date.strftime("%Y-%m-%d"),
end_date.strftime("%Y-%m-%d")
)
hist_data = json.loads(hist_result.text)
assert hist_data["success"] is True
# Get income statement
stmt_result = await get_financial_statements(
symbol,
"income_statement",
"annual"
)
stmt_data = json.loads(stmt_result.text)
assert stmt_data["success"] is True
quote = quote_data["message"]
info = info_data["message"]
hist = hist_data["message"]
stmt = stmt_data["message"]
print(f"\n{'='*60}")
print(f"Complete Analysis for {symbol}")
print(f"{'='*60}")
print(f"Company: {info.get('long_name')}")
print(f"Sector: {info['sector']}")
print(f"Current Price: ${quote['current_price']}")
if "change_percent" in quote:
print(f"Change: {quote['change_percent']}%")
print(f"Historical Data: {hist['total_records']} records")
print(f"Financial Statements: {stmt['total_line_items']} line items")
print(f"{'='*60}\n")
# Run tests
if __name__ == "__main__":
print("=" * 70)
print("Running Yahoo Finance Tools Real API Tests")
print("=" * 70)
print()
# Run with pytest
pytest.main([__file__, "-v", "-s"])
test_youtube_tools.py¶
"""
Real API tests for YouTube transcript extraction.
These tests make actual API calls to YouTube to verify functionality.
"""
import asyncio
import json
import pytest
from pathlib import Path
import sys
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from multimodal_tools import extract_youtube_transcript
class TestYouTubeTranscript:
"""Tests for YouTube transcript extraction."""
@pytest.mark.asyncio
async def test_extract_transcript_by_id(self):
"""Test extracting transcript by video ID."""
# Using a known video with English transcript
# Example: A TED talk or educational video
video_id = "dQw4w9WgXcQ" # A well-known video ID
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["video_id"] == video_id
assert message["language"] == "en"
assert message["total_entries"] > 0
assert len(message["transcript"]) > 0
assert message["full_text_length"] > 0
print(f"✅ Extracted transcript: {message['total_entries']} entries")
print(f" Total text length: {message['full_text_length']} chars")
print(f" First entry: {message['transcript'][0]}")
@pytest.mark.asyncio
async def test_extract_transcript_by_url(self):
"""Test extracting transcript by video URL."""
# Full YouTube URL
video_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_url,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["video_id"] == "dQw4w9WgXcQ"
print(f"✅ Extracted transcript from URL")
print(f" Video ID parsed: {message['video_id']}")
@pytest.mark.asyncio
async def test_extract_transcript_short_url(self):
"""Test extracting transcript by short URL."""
# Short YouTube URL
video_url = "https://youtu.be/dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_url,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert message["video_id"] == "dQw4w9WgXcQ"
print(f"✅ Extracted transcript from short URL")
@pytest.mark.asyncio
async def test_extract_transcript_with_timestamps(self):
"""Test that transcript includes timestamps."""
video_id = "dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
transcript = data["message"]["transcript"]
assert len(transcript) > 0
# Check that entries have timestamps
first_entry = transcript[0]
assert "timestamp" in first_entry
assert "text" in first_entry
# Timestamp should be in MM:SS format
assert ":" in first_entry["timestamp"]
print(f"✅ Transcript has proper timestamps")
print(f" Example: {first_entry['timestamp']} - {first_entry['text'][:50]}")
@pytest.mark.asyncio
async def test_extract_transcript_full_text(self):
"""Test that full text is provided."""
video_id = "dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
assert "full_text" in message
assert len(message["full_text"]) > 0
assert message["full_text_length"] >= len(message["full_text"]) # May or may not be truncated
is_truncated = message["full_text_length"] > len(message["full_text"])
print(f"✅ Full text provided")
print(f" Preview length: {len(message['full_text'])} chars")
print(f" Total length: {message['full_text_length']} chars")
print(f" Truncated: {is_truncated}")
@pytest.mark.asyncio
async def test_extract_transcript_invalid_video(self):
"""Test extracting transcript from invalid video ID."""
result = await extract_youtube_transcript(
video_id="invalid_video_id_xyz",
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is False
assert "error" in data["message"].lower() or "failed" in data["message"].lower()
print("✅ Correctly handled invalid video ID")
@pytest.mark.asyncio
async def test_extract_transcript_metadata(self):
"""Test that proper metadata is included."""
video_id = "dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
metadata = data["metadata"]
assert metadata["video_id"] == video_id
assert metadata["language"] == "en"
assert "translated" in metadata
assert metadata["translated"] is False
print(f"✅ Metadata included")
print(f" Language: {metadata['language']}")
print(f" Translated: {metadata['translated']}")
class TestYouTubeTranscriptTranslation:
"""Tests for transcript translation functionality."""
@pytest.mark.asyncio
async def test_extract_and_translate(self):
"""Test extracting and translating transcript."""
video_id = "dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en",
translate_to_language="es" # Translate to Spanish
)
data = json.loads(result.text)
# Translation might not always work, so handle both cases
if data["success"]:
message = data["message"]
assert message["language"] == "es"
assert data["metadata"]["translated"] is True
print(f"✅ Transcript translated to Spanish")
print(f" Total entries: {message['total_entries']}")
else:
# Translation failed, which is acceptable
print(f"⚠️ Translation not available for this video")
class TestYouTubeTranscriptFormats:
"""Tests for different output formats."""
@pytest.mark.asyncio
async def test_transcript_structure(self):
"""Test the structure of transcript data."""
video_id = "dQw4w9WgXcQ"
result = await extract_youtube_transcript(
video_id=video_id,
language_code="en"
)
data = json.loads(result.text)
assert data["success"] is True
message = data["message"]
# Check structure
assert "video_id" in message
assert "language" in message
assert "transcript" in message
assert "total_entries" in message
assert "full_text" in message
assert "full_text_length" in message
# Check transcript entries structure
if len(message["transcript"]) > 0:
entry = message["transcript"][0]
assert "timestamp" in entry
assert "text" in entry
print(f"✅ Transcript structure validated")
print(f" Fields: {', '.join(message.keys())}")
# Run tests
if __name__ == "__main__":
print("=" * 70)
print("Running YouTube Transcript Tools Real API Tests")
print("=" * 70)
print()
print("Note: These tests use a well-known video ID for testing.")
print("If tests fail, it might be due to YouTube API changes or regional restrictions.")
print()
# Run with pytest
pytest.main([__file__, "-v", "-s"])