48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
import asyncio, sqlite3, json
|
|
from mcp.client.sse import sse_client
|
|
from mcp import ClientSession
|
|
|
|
def get_key(agent):
|
|
con = sqlite3.connect("data/context_hub.db")
|
|
cur = con.cursor()
|
|
cur.execute("SELECT api_key FROM api_keys WHERE agent_name=?", (agent,))
|
|
row = cur.fetchone()
|
|
return row[0] if row else None
|
|
|
|
async def call_as(api_key, tool, args):
|
|
async with sse_client("http://localhost:8000/mcp", headers={"X-API-Key": api_key}) as (read, write):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
return await session.call_tool(tool, args)
|
|
|
|
async def main():
|
|
print("=== TEST 1 : non-regression -- CLAUDE toujours OK sur tt (via table agents) ===")
|
|
r = await call_as(get_key("CLAUDE"), "get_rules", {"scope": "tt"})
|
|
print("OK" if "error" not in r.content[0].text else r.content[0].text)
|
|
|
|
print("=== TEST 2 : non-regression -- GEMINI toujours refuse sur tt ===")
|
|
r2 = await call_as(get_key("GEMINI"), "get_rules", {"scope": "tt"})
|
|
print(r2.content[0].text)
|
|
|
|
print("=== TEST 3 : nouvel agent ajoute SANS redeploiement (insertion directe table agents) ===")
|
|
con = sqlite3.connect("data/context_hub.db")
|
|
cur = con.cursor()
|
|
cur.execute("INSERT OR REPLACE INTO agents (agent_name, agent_type, scopes) VALUES (?, ?, ?)",
|
|
("TEST_AGENT_PHASE5", "test", json.dumps(["coding"])))
|
|
cur.execute("INSERT OR REPLACE INTO api_keys (agent_name, api_key) VALUES (?, ?)",
|
|
("TEST_AGENT_PHASE5", "test-phase5-key-temporaire"))
|
|
con.commit()
|
|
|
|
r3 = await call_as("test-phase5-key-temporaire", "get_rules", {"scope": "coding"})
|
|
print("acces coding:", r3.content[0].text[:100])
|
|
r4 = await call_as("test-phase5-key-temporaire", "get_rules", {"scope": "tt"})
|
|
print("acces tt (doit etre refuse):", r4.content[0].text)
|
|
|
|
print("=== nettoyage ===")
|
|
cur.execute("DELETE FROM api_keys WHERE agent_name='TEST_AGENT_PHASE5'")
|
|
cur.execute("DELETE FROM agents WHERE agent_name='TEST_AGENT_PHASE5'")
|
|
con.commit()
|
|
print("nettoye")
|
|
|
|
asyncio.run(main())
|