#!/usr/bin/env python3
"""OpenViking MCP Server - 使用官方MCP Python SDK"""
import os
import sys
import json
import logging
logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
from mcp.server.fastmcp import FastMCP
try:
import openviking as ov
except ImportError:
print("Error: openviking not found", file=sys.stderr)
sys.exit(1)
mcp = FastMCP("OpenViking")
_client = None
def get_client():
global _client
if _client is None:
data_path = os.environ.get("OPENVIKING_DATA_PATH", "./viking_data")
_client = ov.SyncOpenViking(path=data_path)
_client.initialize()
return _client
@mcp.tool()
def openviking_find(query: str, limit: int = 10) -> str:
"""语义搜索OpenViking知识库
Args:
query: 搜索查询词
limit: 返回结果数量限制
Returns:
JSON格式的搜索结果
"""
try:
client = get_client()
results = client.find(query, target_uri="viking://", limit=limit)
formatted = []
for r in results.resources[:limit]:
try:
preview = client.read(r.uri)[:500]
except:
preview = "Error reading"
formatted.append({"uri": r.uri, "score": round(r.score, 4), "preview": preview})
return json.dumps(formatted, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
@mcp.tool()
def openviking_read(uri: str) -> str:
"""从指定URI读取完整内容
Args:
uri: 资源URI
Returns:
资源内容
"""
try:
return get_client().read(uri)
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def openviking_glob(pattern: str) -> str:
"""使用glob模式匹配文件
Args:
pattern: glob模式（如 **/*.pdf）
Returns:
匹配文件列表
"""
try:
results = get_client().glob(pattern, uri="viking://")
return json.dumps(results, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
@mcp.tool()
def openviking_abstract(uri: str) -> str:
"""获取资源的L0摘要
Args:
uri: 资源URI
Returns:
摘要内容
"""
try:
return get_client().abstract(uri)
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def openviking_overview(uri: str) -> str:
"""获取资源的L1概览
Args:
uri: 资源URI
Returns:
概览内容
"""
try:
return get_client().overview(uri)
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def openviking_grep(pattern: str, case_insensitive: bool = False) -> str:
"""全文内容搜索
Args:
pattern: 搜索模式
case_insensitive: 是否忽略大小写
Returns:
搜索结果
"""
try:
results = get_client().grep("viking://", pattern, case_insensitive=case_insensitive)
return json.dumps(results, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
if __name__ == "__main__":
mcp.run(transport="stdio")
