Spaces:
Sleeping
Sleeping
File size: 966 Bytes
c16e1c9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
import os
import httpx
from dotenv import load_dotenv
load_dotenv()
class AdminClient:
"""
Communicates with the Admin MCP governance server.
"""
def __init__(self):
self.base_url = os.getenv("ADMIN_MCP_URL")
self.alert_endpoint = f"{self.base_url}/alert"
async def alert(self, tenant_id: str, message: str, redflag: dict):
"""
Sends redflag violation to admin server.
"""
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.alert_endpoint,
json={
"tenant_id": tenant_id,
"message": message,
"violations": redflag.get("matches", []),
}
)
return response.status_code == 200
except Exception as e:
print("Admin Client Error:", e)
return False
|