38 lines
1.7 KiB
Python
38 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
import json, os, sys
|
|
os.environ.setdefault('MCP_DB','/data/loogle_mcp.db')
|
|
os.environ.setdefault('MCP_VECTOR_FALLBACK','/data/vector_fallback.db')
|
|
from app.db import init_db, get_conn
|
|
from app.knowledge import gitea_indexer
|
|
from app.mcp import tools
|
|
init_db()
|
|
assert get_conn().execute("SELECT 1 FROM sqlite_master WHERE name='indexed_gitea_files'").fetchone()
|
|
repo='daniele/rete'
|
|
for path in ['ha/FAILOVER-VOLUMES.md','ha/RUNBOOK-failover.md']:
|
|
r=gitea_indexer.index_file(repo, path, username='daniele', private=True, force=True)
|
|
print('indexed', path, r.get('chunks'), r.get('skipped'), flush=True)
|
|
files=gitea_indexer.list_indexed_files(limit=10, repo=repo)
|
|
print('db_files', [f['path'] for f in files], flush=True)
|
|
for q in ('failover tier-b','runbook'):
|
|
hits=gitea_indexer.search_gitea_knowledge('daniele', q, limit=3, is_admin=True)
|
|
print('search', q, hits[0]['path'] if hits else None, round(float(hits[0]['score']),3) if hits else None, flush=True)
|
|
if not hits: sys.exit(1)
|
|
claims={'sub':'daniele','scope':'knowledge:read gitea:read admin'}
|
|
for tool,args in [
|
|
('search_gitea_knowledge',{'query':'failover tier-b','limit':5}),
|
|
('list_gitea_indexed_files',{'repo':repo,'limit':10}),
|
|
('reindex_gitea_repo',{'repo':repo,'max_files':1}),
|
|
('search_knowledge',{'query':'failover keepalived','limit':8}),
|
|
]:
|
|
out=tools.call_tool(tool,args,claims)
|
|
p=json.loads(out['content'][0]['text'])
|
|
if tool=='search_knowledge':
|
|
src=sorted({r.get('source') for r in p['results']})
|
|
print(tool, src, flush=True)
|
|
assert 'gitea' in src
|
|
elif tool=='list_gitea_indexed_files':
|
|
print(tool, p['count'], flush=True)
|
|
else:
|
|
print(tool, 'ok', flush=True)
|
|
print('OK P3 COMPLETE', flush=True)
|