bolbol/context-hub
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(context-hub): filtrage per-agent scope infra (bruit mcp-nas non pertinent pour Gemini/Hermes) + purge cle obsolete mcp_nas_heredoc_piege_confirme

Browse Source
This commit is contained in:
bolbol
2026-08-16 21:35:05 +00:00
parent 39ee17dcac
commit 210669ef00
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/mcp.py
+5 -4
View File
@@ -11,7 +11,7 @@ except ImportError:
from mcp.types import Tool, TextContent
from app.auth import get_current_agent
from app.scopes import has_scope_access, ALL_SCOPES
from app.models import (get_rule, get_all_rules, get_all_ports, get_port, set_rule, get_agent_by_key, update_key_last_used, add_memory_entry, list_memory_entries)
from app.models import (get_rule, get_all_rules, get_all_ports, get_port, set_rule, get_agent_by_key, update_key_last_used, add_memory_entry, list_memory_entries, filter_content_for_agent)
from app.git_sync import push_scope_markdown
logger = logging.getLogger(__name__)
@@ -141,7 +141,7 @@ async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
rule = await get_rule(scope)
if not rule:
return [TextContent(type="text", text=json.dumps({"error": "Scope not found"}))]
return [TextContent(type="text", text=json.dumps(rule))]
return [TextContent(type="text", text=json.dumps(filter_content_for_agent(rule, caller_agent)))]
elif name == "get_ports":
if not has_scope_access(caller_agent, "infra"):
@@ -156,6 +156,7 @@ async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
for scope, content_ in all_rules.items():
if not has_scope_access(caller_agent, scope):
continue
content_ = filter_content_for_agent(content_, caller_agent)
if query in json.dumps(content_).lower():
results.append({"scope": scope, "content": content_})
return [TextContent(type="text", text=json.dumps(results))]
@@ -165,7 +166,7 @@ async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
config = {}
for scope in ALL_SCOPES:
if has_scope_access(caller_agent, scope):
config[scope] = all_rules.get(scope, {})
config[scope] = filter_content_for_agent(all_rules.get(scope, {}), caller_agent)
return [TextContent(type="text", text=json.dumps(config))]
elif name == "check_port":
@@ -183,7 +184,7 @@ async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
result = {}
for scope in ALL_SCOPES:
if has_scope_access(caller_agent, scope):
result[scope] = all_rules.get(scope, {})
result[scope] = filter_content_for_agent(all_rules.get(scope, {}), caller_agent)
return [TextContent(type="text", text=json.dumps({"agent": caller_agent, "scopes": result}))]
elif name == "update_rule":