r/LangChain 的一篇文章记录了一个自治 DaaS 架构：Google Dorks 发现、Llama-3 转换、带有 SQLite 缓存的 MCP 服务。本教程在 Scavio 上采用相同的架构。
前置条件
- Python 3.10+
- 浪链
- Scavio API 密钥
- SQLite（内置）
操作指南
步骤 1: 呆子名单
定义发现查询。
Python
DORKS = [
'site:gov.br filetype:pdf 2026 contratos',
'site:europa.eu filetype:pdf AI Act',
'site:sec.gov filetype:pdf 10-K 2026',
]步骤 2: 通过 Scavio /search 发现
运行每个傻瓜。
Python
import os, requests
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def discover(q):
return requests.post('https://api.scavio.dev/api/v1/search', headers=H, json={'query': q}).json()步骤 3: 通过 /extract 提取 PDF
每个发现的 URL。
Python
def fetch(url):
return requests.post('https://api.scavio.dev/api/v1/extract', headers=H, json={'url': url, 'format': 'markdown'}).json()步骤 4: 法学硕士转型
Llama-3（或任何 LLM）将 markdown 转换为类型化 JSON。
Python
# Prompt: 'Extract a strict JSON: {title, jurisdiction, deadline, summary, risk_level}.'
# Use Groq for cheap Llama-3, or Anthropic Sonnet for quality.步骤 5: SQLite缓存层
不到 50 毫秒的重复查找。
Python
import sqlite3, json, time
conn = sqlite3.connect('daas.db')
conn.execute('CREATE TABLE IF NOT EXISTS items(url TEXT PRIMARY KEY, payload TEXT, ts REAL)')
def cache_set(url, payload):
conn.execute('INSERT OR REPLACE INTO items VALUES (?, ?, ?)', (url, json.dumps(payload), time.time()))
conn.commit()步骤 6: 通过MCP为下游代理服务
将缓存包装在 FastMCP 服务器中。
Python
# from fastmcp import FastMCP
# mcp = FastMCP('daas')
# @mcp.tool()
# def get_item(url: str) -> dict:
# row = conn.execute('SELECT payload FROM items WHERE url=?', (url,)).fetchone()
# return json.loads(row[0]) if row else {}Python 示例
Python
# Wrap discover + fetch + transform + cache in a daily cron.
# Downstream CrewAI / LangChain agents query the MCP for sub-50ms typed JSON.JavaScript 示例
JavaScript
// Same architecture in TS with better-sqlite3 and the MCP TS SDK.预期输出
JSON
Daily 4 AM cron pulls dorks, fetches PDFs, transforms to typed JSON, caches in SQLite. Downstream agents read from cache in 50ms instead of running real-time scrapers.