Return Mattermost permalinks from search MCP using config host.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+68
-2
@@ -40,6 +40,7 @@ Requires: pip install fastmcp
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import json
|
||||
import os
|
||||
from typing import Any, Literal, Optional
|
||||
|
||||
@@ -52,6 +53,9 @@ mcp = FastMCP("mattermost-search")
|
||||
|
||||
SearchMode = Literal["message", "file_name", "file_content"]
|
||||
|
||||
# Public Mattermost site for permalinks. Override with MM_SITE_URL or config.json "host".
|
||||
_DEFAULT_SITE_HOST = "mattermost.eltex-co.ru"
|
||||
|
||||
|
||||
def _env_root() -> str:
|
||||
return os.environ.get("MM_SEARCH_ROOT", "results")
|
||||
@@ -61,6 +65,52 @@ def _env_db() -> str:
|
||||
return os.environ.get("MM_SEARCH_DB", "mattermost_index.sqlite")
|
||||
|
||||
|
||||
def _env_config_path() -> str:
|
||||
return os.environ.get("MM_CONFIG_PATH", "config.json")
|
||||
|
||||
|
||||
def _site_url() -> str:
|
||||
"""https://mattermost.eltex-co.ru — from MM_SITE_URL, else config.json host."""
|
||||
env = (os.environ.get("MM_SITE_URL") or "").strip().rstrip("/")
|
||||
if env:
|
||||
if "://" not in env:
|
||||
return f"https://{env}"
|
||||
return env
|
||||
path = _env_config_path()
|
||||
if os.path.isfile(path):
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
host = (json.load(f).get("host") or "").strip()
|
||||
if host:
|
||||
if "://" in host:
|
||||
return host.rstrip("/")
|
||||
return f"https://{host}"
|
||||
except (OSError, ValueError, TypeError):
|
||||
pass
|
||||
return f"https://{_DEFAULT_SITE_HOST}"
|
||||
|
||||
|
||||
def _permalink(match: dict[str, Any], site: str) -> Optional[str]:
|
||||
team = (match.get("team") or "").strip()
|
||||
post = match.get("post") if isinstance(match.get("post"), dict) else {}
|
||||
post_id = (post.get("id") or "").strip()
|
||||
if not team or not post_id:
|
||||
return None
|
||||
return f"{site}/{team}/pl/{post_id}"
|
||||
|
||||
|
||||
def _enrich_matches(matches: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
site = _site_url()
|
||||
out: list[dict[str, Any]] = []
|
||||
for m in matches:
|
||||
item = dict(m)
|
||||
link = _permalink(item, site)
|
||||
if link:
|
||||
item["permalink"] = link
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
|
||||
def _parse_date(value: str) -> dt.datetime:
|
||||
try:
|
||||
return mm_search_lib.parse_iso_date(value)
|
||||
@@ -170,7 +220,12 @@ def _format_human(matches: list[dict[str, Any]], show_file: bool) -> str:
|
||||
channel = m.get("channel") or "?"
|
||||
team = m.get("team") or ""
|
||||
team_prefix = f"[{team}]" if team else ""
|
||||
lines.append(f"{prefix}{team_prefix}[{channel}] {created} {username}: {msg}{attachment_suffix}")
|
||||
link = m.get("permalink") or ""
|
||||
link_suffix = f" {link}" if link else ""
|
||||
lines.append(
|
||||
f"{prefix}{team_prefix}[{channel}] {created} {username}: {msg}"
|
||||
f"{attachment_suffix}{link_suffix}"
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@@ -195,6 +250,8 @@ def mm_search(
|
||||
"""Search Mattermost export JSON files produced by mattermost-dl.py.
|
||||
|
||||
Mirrors mm_search.py CLI. Returns structured matches or human-readable lines.
|
||||
Each match includes permalink https://mattermost.eltex-co.ru/{team}/pl/{post.id}
|
||||
(host from MM_SITE_URL or config.json). Cite that URL when sharing a post.
|
||||
|
||||
Args:
|
||||
query: Search string or regex (when regex=True).
|
||||
@@ -233,6 +290,7 @@ def mm_search(
|
||||
db=resolved_db,
|
||||
no_index=no_index,
|
||||
)
|
||||
matches = _enrich_matches(matches)
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"count": len(matches),
|
||||
@@ -242,6 +300,7 @@ def mm_search(
|
||||
"search_mode": search_mode,
|
||||
"root": resolved_root,
|
||||
"db": resolved_db,
|
||||
"site_url": _site_url(),
|
||||
}
|
||||
|
||||
if output == "human":
|
||||
@@ -331,6 +390,7 @@ def mm_search_tag(
|
||||
|
||||
Plain SQL substring match — no FTS/rank. Use for daily tags like plan/report
|
||||
when you have after/before (e.g. one day or one week).
|
||||
Each match includes permalink https://mattermost.eltex-co.ru/{team}/pl/{post.id}.
|
||||
|
||||
Args:
|
||||
keyword: Tag without or with leading # (e.g. "plan", "report", "#plan").
|
||||
@@ -361,6 +421,7 @@ def mm_search_tag(
|
||||
limit=limit,
|
||||
)
|
||||
)
|
||||
matches = _enrich_matches(matches)
|
||||
tag = keyword.strip()
|
||||
if not tag.startswith("#"):
|
||||
tag = f"#{tag}"
|
||||
@@ -372,6 +433,7 @@ def mm_search_tag(
|
||||
"after": after,
|
||||
"before": before,
|
||||
"db": resolved_db,
|
||||
"site_url": _site_url(),
|
||||
"matches": matches,
|
||||
}
|
||||
|
||||
@@ -381,7 +443,10 @@ def mm_search_index_status(
|
||||
db: Optional[str] = None,
|
||||
root: Optional[str] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Report whether the sqlite index and export root are available."""
|
||||
"""Report whether the sqlite index and export root are available.
|
||||
|
||||
Also returns site_url (https://mattermost.eltex-co.ru) for building permalinks.
|
||||
"""
|
||||
resolved_db = db or _env_db()
|
||||
resolved_root = root or _env_root()
|
||||
db_exists = os.path.exists(resolved_db)
|
||||
@@ -410,6 +475,7 @@ def mm_search_index_status(
|
||||
"json_files": json_count,
|
||||
"users": user_count,
|
||||
"index_required_for": ["file_name", "file_content"],
|
||||
"site_url": _site_url(),
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user