其实我一直都想做一下，哈哈哈，刚好这个点，我们先看看最关键的一点
1.其实我今天再看skill中关注到其中一点有一个web search 功能，哈哈哈，大家能猜到了对吧
我试图想要去建一个MCP功能；
这个工作流：
1.建一个临时工作库
2.开始搜索关键词内容
3，开始结合关键词得到的内容来整合，但不上传内容
4.随后清理临时知识库
5.Agent开始执行
2.其实整个过程使用iflow cli来配置setting.json 利用的硅基流动API功能。
目标就是实现web search 和 web fetch功能嘛
哈哈哈，只是
一直显示无法使用真的很可惜，arg 参数设定到那个需要的脚本：
我喜欢用.py 经过规划和分析仅仅只需要实现web search功能，因此我没有想那么多
虽然没有办法使用哈
脚本：
#!/usr/bin/env python3
“”"
iflow Web Search MCP 服务器（简化版）
仅实现联网搜索功能，不导入知识库，直接返回结果供 Agent 使用
安装依赖:
pip install mcp
配置:
在 settings.json 的 “mcpServers” 中添加:
{
“iflow-web-search”: {
“command”: “python”,
“args”: [“D:\iflow_nb\iflow_web_search_mcp_simple.py”],
“env”: {
“IFLOW_API_KEY”: “your_api_key_here”
},
“disabled”: false,
“autoApprove”: [“web_search”]
}
}
“”"
import asyncio
import json
import os
import sys
import time
from typing import Any, Dict, List, Optional
import requests
尝试导入 mcp 库
try:
from mcp.server import Server, InitializationOptions, NotificationOptions
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
except ImportError:
print(“错误: 未安装 mcp 库。请运行: pip install mcp”, file=sys.stderr)
sys.exit(1)
将 iflow-nb 脚本目录加入路径
sys.path.insert(0, os.path.join(os.path.dirname(file), “.gemini”, “skills”, “iflow-nb”, “scripts”))
尝试导入 iflow_common
try:
from iflow_common import load_credentials, start_search, poll_search, log # type: ignore
except ImportError:
# 如果导入失败，定义简化版本
def load_credentials():
“”“加载 iflow API 凭证”“”
key = os.environ.get(“IFLOW_API_KEY”, “”)
if not key:
config_path = os.path.expanduser(“~/.config/iflow-nb/api-key”)
if os.path.exists(config_path):
with open(config_path, “r”) as f:
key = f.read().strip()
if not key:
raise ValueError(“IFLOW_API_KEY 未配置”)
base_url = os.environ.get(“IFLOW_BASE_URL”, “https://platform.iflow.cn”)
return key, base_url
def log(msg):
print(f">>> {msg}", file=sys.stderr)
配置
BASE_URL = “https://platform.iflow.cn”
创建 MCP 服务器
server = Server(“iflow-web-search-simple”)
定义工具
TOOLS = [
Tool(
name=“web_search”,
description=“联网搜索功能。使用 iflow API 进行网络搜索，返回搜索结果供 Agent 分析和提取关键内容。支持快速搜索(FAST_SEARCH，约4秒)和深度研究(DEEP_RESEARCH，约2-5分钟)。”,
inputSchema={
“type”: “object”,
“properties”: {
“query”: {
“type”: “string”,
“description”: “搜索关键词（必填）”
},
“search_type”: {
“type”: “string”,
“enum”: [“FAST_SEARCH”, “DEEP_RESEARCH”],
“description”: “搜索类型：FAST_SEARCH（快速，约4秒）或 DEEP_RESEARCH（深度，约2-5分钟）”,
“default”: “FAST_SEARCH”
},
“max_results”: {
“type”: “integer”,
“description”: “返回的最大结果数量（0=全部）”,
“default”: 10
}
},
“required”: [“query”]
}
)
]
@server.list_tools()
async def handle_list_tools() → List[Tool]:
“”“列出可用工具”“”
return TOOLS
@server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any]) → List[TextContent]:
“”“处理工具调用”“”
if name == “web_search”:
return await handle_web_search(**arguments)
else:
raise ValueError(f"未知工具: {name}")
async def handle_web_search(
query: str,
search_type: str = “FAST_SEARCH”,
max_results: int = 10
) → List[TextContent]:
“”“执行联网搜索”“”
try:
# 1. 验证凭证
try:
api_key, base_url = load_credentials()
except ValueError as e:
return [TextContent(
type=“text”,
text=f"错误: {str(e)}\n\n请设置 IFLOW_API_KEY 环境变量或创建 ~/.config/iflow-nb/api-key 文件"
)]
headers = {"Authorization": f"Bearer {api_key}"}
# 2. 创建临时知识库
temp_kb_name = f"temp_search_{int(time.time())}"
log(f"创建临时知识库: {temp_kb_name}")
resp = requests.post(
f"{base_url}/api/v1/knowledge/createNotebook",
headers=headers,
json={"name": temp_kb_name, "description": "临时搜索知识库"}
)
if not resp.json().get("success"):
return [TextContent(type="text", text="创建临时知识库失败")]
kb_id = resp.json().get("data", {}).get("id")
try:
# 3. 发起搜索
log(f"发起搜索: {query} ({search_type})")
body = {
"query": query,
"type": search_type,
"source": "WEB",
"notebookId": kb_id
}
resp = requests.post(
f"{base_url}/api/v1/knowledge/startSearch",
headers=headers,
json=body
)
if not resp.json().get("success"):
code = resp.json().get("code", "")
if code == "40010":
return [TextContent(type="text", text="错误: 深度研究并发限制，已有一个深度研究任务正在处理中，请稍后再试")]
return [TextContent(type="text", text=f"搜索发起失败: {resp.json().get('message', '未知错误')}")]
search_id = resp.json().get("data", "")
# 4. 轮询搜索结果
if search_type == "FAST_SEARCH":
max_wait, interval = 60, 3
else: # DEEP_RESEARCH
max_wait, interval = 600, 10
log(f"轮询搜索结果 (最多等待 {max_wait} 秒)...")
elapsed = 0
search_data = None
while elapsed < max_wait:
time.sleep(interval)
elapsed += interval
resp = requests.get(
f"{base_url}/api/v1/knowledge/getSearchResult?notebookId={kb_id}&id={search_id}",
headers=headers
)
data = resp.json().get("data", {})
status = data.get("status", "unknown")
if status == "completed":
search_data = data
break
elif status in ("failed", "dismissed"):
return [TextContent(type="text", text=f"搜索失败: status={status}")]
if not search_data:
return [TextContent(type="text", text="搜索超时")]
# 5. 处理结果
results = search_data.get("results", [])
result_count = len(results)
if max_results > 0:
results = results[:max_results]
# 6. 构建返回文本
result_text = f"""# 联网搜索结果
搜索信息
- 查询词: {query}
- 搜索类型: {search_type}
- 总结果数: {result_count}
- 返回结果数: {len(results)}
搜索结果
“”"
for i, item in enumerate(results, 1):
title = item.get("title", "无标题")
url = item.get("url", "")
snippet = item.get("snippet", item.get("content", "无摘要"))
content_type = item.get("contentType", "WEBSITE")
score = item.get("score")
result_text += f"### {i}. {title}\n"
result_text += f"- **URL**: {url}\n"
result_text += f"- **类型**: {content_type}\n"
if score is not None:
result_text += f"- **相关性**: {score:.2f}\n"
result_text += f"- **摘要**: {snippet}\n\n"
return [TextContent(type="text", text=result_text)]
finally:
# 7. 删除临时知识库
log(f"删除临时知识库: {kb_id}")
requests.post(
f"{base_url}/api/v1/knowledge/deleteNotebook",
headers=headers,
json={"id": kb_id}
)
except Exception as e:
return [TextContent(
type="text",
text=f"搜索过程中发生错误: {str(e)}\n\n请检查:\n1. IFLOW_API_KEY 是否正确配置\n2. 网络连接是否正常\n3. API 服务是否可用"
)]
async def main():
“”“启动 MCP 服务器”“”
# 检查凭证
try:
load_credentials()
log(“凭证验证成功”)
except ValueError as e:
log(f"凭证错误: {e}")
log(“请设置 IFLOW_API_KEY 环境变量或创建 ~/.config/iflow-nb/api-key 文件”)
sys.exit(1)
# 启动服务器
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="iflow-web-search-simple",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
if name == “main”:
asyncio.run(main())
但是这是一个思路哈
这种skill 拆分，很好的实现功能获取，功能拆解，功能获取，定制功能的功能
好啦，我们直接到核心的pipeline 6 的伪代码分享吧；
Pipeline 6: 网络搜索并导入知识库
功能说明
进行网络搜索，将搜索结果导入知识库，可选择生成内容。
命令示例
python pipeline_web_search.py --name “搜索结果库” --query “Python教程” --num 10 --output-type PDF
python pipeline_web_search.py --search-only --query “天气”
伪代码流程
步骤 1: 接收参数
- –name: 知识库名称（必填，如果不使用–search-only）
- –query: 搜索查询（必填）
- –num: 搜索结果数量（默认10）
- –output-type: 输出类型（默认PDF）
- –query-for-generate: 创作的查询要求（可选）
- –search-only: 仅搜索不导入（可选）
- –poll-creation: 等待创作完成（可选）
步骤 2: 网络搜索
- 调用API: POST /api/v1/search
- 传递参数: query, num
- 获取搜索结果列表
步骤 3: 输出搜索结果
- 打印每个结果: 标题、URL、摘要
步骤 4: 导入知识库（如果不使用–search-only）
如果使用–search-only，跳过此步骤
-
创建知识库（如果–name提供）:
- 调用API: POST /api/v1/knowledge/saveCollection
- 获取collectionId
-
导入搜索结果:
- 遍历搜索结果
- 调用api_upload上传URL
- 收集contentId
-
轮询解析状态:
- 循环直到解析完成或超时(300秒)
步骤 5: 生成内容（如果设置–output-type且非–search-only）
- 调用submit_creation提交创作任务
- 如果设置–poll-creation，轮询创作状态
核心API
api_post, api_upload, poll_parsing, submit_creation, poll_creation
但是通过步骤1-3可以看出其实本skill功能即可实现网络搜索功能，对吧，哈哈哈很有意思
就是容易出现
哈哈哈，这个问题可以等待许愿神灯呀
。