mirror of
https://github.com/invariantlabs-ai/invariant-gateway.git
synced 2026-05-15 20:48:17 +02:00
Luca Beurer-Kellner
e18c6b5bdb
* use select() before readline * support for setting static metadata for MCP sessions * nest extra mcp metadata in metadata object * unify session metadata * extra metadata tests * use empty object as parameters, if None * list_tools as tool call * offset indices in tests * test: adjust addresses * mcp: make error reporting configurable * line logging * log version * verbose logging + loud exception failure * add server and client name to policy get * append trace even if not pushing * port tools/list message support to SSE * use python -m build * adjust guardrail failure address * support for blocking tools/list in SSE * use error-based failure response format by default * tools/list test * don't list_tools in stdio connect * flaky test: handle second possible result in anthropic streaming case --------- Co-authored-by: knielsen404 <kristian@invariantlabs.ai>
30 lines
1007 B
Python
30 lines
1007 B
Python
"""Cusym log configuration."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
from builtins import print as builtins_print
|
|
|
|
os.makedirs(os.path.join(os.path.expanduser("~"), ".invariant"), exist_ok=True)
|
|
MCP_LOG_FILE = open(
|
|
os.path.join(os.path.expanduser("~"), ".invariant", "mcp.log"),
|
|
"a",
|
|
buffering=1,
|
|
)
|
|
sys.stderr = MCP_LOG_FILE
|
|
|
|
|
|
def mcp_log(*args, **kwargs) -> None:
|
|
"""Custom print function to redirect output to log_out."""
|
|
builtins_print(*args, **kwargs, file=MCP_LOG_FILE, flush=True)
|
|
|
|
def format_errors_in_response(errors: list[dict]) -> str:
|
|
"""Format a list of errors in a response string."""
|
|
|
|
def format_error(error: dict) -> str:
|
|
msg = " ".join(error.get("args", []))
|
|
msg += " ".join([f"{k}={v}" for k, v in error.get("kwargs", {}).items()])
|
|
msg += f" ([{error.get('guardrail', {}).get('id', 'unknown-guardrail')}] {error.get('guardrail', {}).get('name', 'unknown guardrail')})"
|
|
return msg
|
|
|
|
return ", ".join([format_error(error) for error in errors]) |