test: onboarding HERMES_NABIL (acces complet)
This commit is contained in:
@@ -0,0 +1,32 @@
|
|||||||
|
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,))
|
||||||
|
return cur.fetchone()[0]
|
||||||
|
|
||||||
|
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():
|
||||||
|
key = get_key("HERMES_NABIL")
|
||||||
|
|
||||||
|
print("=== get_my_context ===")
|
||||||
|
r = await call_as(key, "get_my_context", {"agent": "HERMES_NABIL"})
|
||||||
|
print(sorted(json.loads(r.content[0].text)["scopes"].keys()))
|
||||||
|
|
||||||
|
print("=== acces tt (doit reussir, acces complet) ===")
|
||||||
|
r2 = await call_as(key, "get_rules", {"scope": "tt"})
|
||||||
|
print("OK" if "error" not in r2.content[0].text else r2.content[0].text)
|
||||||
|
|
||||||
|
print("=== acces coding via get_context_pack (doit reussir) ===")
|
||||||
|
r3 = await call_as(key, "get_context_pack", {"scope": "coding"})
|
||||||
|
print(r3.content[0].text)
|
||||||
|
|
||||||
|
asyncio.run(main())
|
||||||
Reference in New Issue
Block a user