51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
import asyncio, sqlite3
|
|
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,))
|
|
return cur.fetchone()[0]
|
|
|
|
async def call_as(agent, tool, args):
|
|
key = get_key(agent)
|
|
async with sse_client("http://localhost:8000/mcp", headers={"X-API-Key": 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 : CLAUDE record_lesson (coding, common-error) ===")
|
|
r = await call_as("CLAUDE", "record_lesson", {
|
|
"scope": "coding", "type": "common-error",
|
|
"title": "Test phase 3 - a supprimer",
|
|
"body": "Entree de test pour valider record_lesson/get_context_pack.",
|
|
"project": "context-hub", "tags": ["test"]
|
|
})
|
|
print(r.content[0].text)
|
|
|
|
print("=== TEST 2 : GEMINI record_lesson (coding, best-practice) ===")
|
|
r2 = await call_as("GEMINI", "record_lesson", {
|
|
"scope": "coding", "type": "best-practice",
|
|
"title": "Test phase 3 bis - a supprimer",
|
|
"body": "Deuxieme entree de test, agent different.",
|
|
})
|
|
print(r2.content[0].text)
|
|
|
|
print("=== TEST 3 : HERMES_TT tente record_lesson (coding) -- doit etre refuse (pas de scope coding) ===")
|
|
r3 = await call_as("HERMES_TT", "record_lesson", {
|
|
"scope": "coding", "type": "decision", "title": "x", "body": "y"
|
|
})
|
|
print(r3.content[0].text)
|
|
|
|
print("=== TEST 4 : get_context_pack(coding) -- doit contenir les 2 entrees ===")
|
|
r4 = await call_as("CLAUDE", "get_context_pack", {"scope": "coding"})
|
|
print(r4.content[0].text)
|
|
|
|
print("=== TEST 5 : get_context_pack(coding, project=context-hub) -- doit filtrer a 1 entree ===")
|
|
r5 = await call_as("CLAUDE", "get_context_pack", {"scope": "coding", "project": "context-hub"})
|
|
print(r5.content[0].text)
|
|
|
|
asyncio.run(main())
|