Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix mcp search

Browse Source
This commit is contained in:
larssand
2026-06-21 22:18:58 +02:00
parent 5010e53e0a
commit 35c21332f3
2 changed files with 19 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/fgai/graylog_source.py
View File

@@ -44,8 +44,17 @@ class GraylogStreamSource:
def fetch(self) -> tuple[list[LogEvent], dict[str, object]]:
status = self.client.probe()
result = self.client.call_tool("search_messages", {"query": self.query, "stream": self.stream, "limit": 1000})
arguments: dict[str, object] = {"query": self.query, "limit": 1000, "range_seconds": 300}
if self.stream:
arguments["streams"] = [self.stream]
result = self.client.call_tool("search_messages", arguments)
content = result.get("result", {}).get("content", []) if isinstance(result.get("result"), dict) else []
if isinstance(result.get("result"), dict) and result["result"].get("isError"):
detail = next(
(str(item.get("text")) for item in content if isinstance(item, dict) and item.get("type") == "text"),
"Graylog search failed",
)
raise RuntimeError(f"graylog_search_error: {detail}")
records: list[dict[str, object]] = []
for item in content if isinstance(content, list) else []:
if isinstance(item, dict) and item.get("type") == "text":