perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

fix: use StdioServerParameters to bypass Agno executable whitelist

Browse Source 
Agno's MCPTools has an undocumented executable whitelist that blocks
gitea-mcp (Go binary). Switch to server_params=StdioServerParameters()
which bypasses this restriction. Also fixes:

- Use tools.session.call_tool() for standalone invocation (MCPTools
  doesn't expose call_tool() directly)
- Use close() instead of disconnect() for cleanup
- Resolve gitea-mcp path via ~/go/bin fallback when not on PATH
- Stub mcp.client.stdio in test conftest

Smoke-tested end-to-end against real Gitea: connect, list_issues,
create issue, close issue, create_gitea_issue_via_mcp — all pass.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Trip T
2026-03-12 22:03:45 -04:00
parent 8aef55ac07
commit bd1aa55904
3 changed files with 143 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
src/timmy/mcp_tools.py
View File

@@ -21,6 +21,8 @@ Usage::
from __future__ import annotations
import logging
import os
import shutil
import sqlite3
import uuid
from datetime import datetime
@@ -34,11 +36,55 @@ logger = logging.getLogger(__name__)
_issue_session = None
def _parse_command(command_str: str) -> tuple[str, list[str]]:
"""Split a command string into (executable, args).
Handles ``~/`` expansion and resolves via PATH if needed.
E.g. ``"gitea-mcp -t stdio"`` → ``("/Users/x/go/bin/gitea-mcp", ["-t", "stdio"])``
"""
parts = command_str.split()
executable = os.path.expanduser(parts[0])
# If not an absolute path, resolve via shutil.which
if not os.path.isabs(executable):
resolved = shutil.which(executable)
if resolved:
executable = resolved
else:
# Check common binary locations not always on PATH
for candidate_dir in ["~/go/bin", "~/.local/bin", "~/bin"]:
candidate = os.path.expanduser(os.path.join(candidate_dir, parts[0]))
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
executable = candidate
break
return executable, parts[1:]
def _gitea_server_params():
"""Build ``StdioServerParameters`` for the Gitea MCP server."""
from mcp.client.stdio import StdioServerParameters
exe, args = _parse_command(settings.mcp_gitea_command)
return StdioServerParameters(
command=exe,
args=args,
env={
"GITEA_ACCESS_TOKEN": settings.gitea_token,
"GITEA_HOST": settings.gitea_url,
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
},
)
def create_gitea_mcp_tools():
"""Create an MCPTools instance for the Gitea MCP server.
Returns None if Gitea is disabled or not configured (no token).
The returned MCPTools is lazy — Agno connects it on first ``arun()``.
Uses ``server_params`` instead of ``command`` to bypass Agno's
executable whitelist (gitea-mcp is a Go binary not in the list).
"""
if not settings.gitea_enabled or not settings.gitea_token:
logger.debug("Gitea MCP: disabled or no token configured")
@@ -47,15 +93,8 @@ def create_gitea_mcp_tools():
try:
from agno.tools.mcp import MCPTools
# Build command — gitea-mcp expects "-t stdio" for stdio transport
command = settings.mcp_gitea_command
tools = MCPTools(
command=command,
env={
"GITEA_ACCESS_TOKEN": settings.gitea_token,
"GITEA_HOST": settings.gitea_url,
},
server_params=_gitea_server_params(),
include_tools=[
"issue_write",
"issue_read",
@@ -80,14 +119,28 @@ def create_filesystem_mcp_tools():
Returns None if the command is not configured.
Scoped to the project repo_root directory.
Uses ``server_params`` for consistency (npx is whitelisted by Agno
but server_params is the more robust approach).
"""
try:
from agno.tools.mcp import MCPTools
from mcp.client.stdio import StdioServerParameters
command = f"{settings.mcp_filesystem_command} {settings.repo_root}"
# Parse the base command, then append repo_root as an extra arg
exe, args = _parse_command(settings.mcp_filesystem_command)
args.append(str(settings.repo_root))
params = StdioServerParameters(
command=exe,
args=args,
env={
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
},
)
tools = MCPTools(
command=command,
server_params=params,
include_tools=[
"read_file",
"write_file",
@@ -151,6 +204,9 @@ async def create_gitea_issue_via_mcp(title: str, body: str = "", labels: str = "
Used by the thinking engine's ``_maybe_file_issues()`` post-hook.
Manages its own MCPTools session with lazy connect + graceful failure.
Uses ``tools.session.call_tool()`` for direct MCP invocation — the
``MCPTools`` wrapper itself does not expose ``call_tool()``.
Args:
title: Issue title.
body: Issue body (markdown).
@@ -169,11 +225,7 @@ async def create_gitea_issue_via_mcp(title: str, body: str = "", labels: str = "
if _issue_session is None:
_issue_session = MCPTools(
command=settings.mcp_gitea_command,
env={
"GITEA_ACCESS_TOKEN": settings.gitea_token,
"GITEA_HOST": settings.gitea_url,
},
server_params=_gitea_server_params(),
timeout_seconds=settings.mcp_timeout,
)
@@ -200,8 +252,8 @@ async def create_gitea_issue_via_mcp(title: str, body: str = "", labels: str = "
"body": full_body,
}
# Call the MCP tool directly via the session
result = await _issue_session.call_tool("issue_write", arguments=args)
# Call via the underlying MCP session (MCPTools doesn't expose call_tool)
result = await _issue_session.session.call_tool("issue_write", arguments=args)
# Bridge to local work order
label_list = [tag.strip() for tag in labels.split(",") if tag.strip()] if labels else []
@@ -221,7 +273,7 @@ async def close_mcp_sessions() -> None:
global _issue_session
if _issue_session is not None:
try:
await _issue_session.disconnect()
await _issue_session.close()
except Exception as exc:
logger.debug("MCP session disconnect error: %s", exc)
logger.debug("MCP session close error: %s", exc)
_issue_session = None

2
tests/conftest.py
View File

@@ -20,6 +20,8 @@ except ImportError:
for _mod in [
"airllm",
"mcp",
"mcp.client",
"mcp.client.stdio",
"mcp.registry",
"telegram",
"telegram.ext",

95
tests/timmy/test_mcp_tools.py
View File

@@ -6,12 +6,42 @@ import pytest
from timmy.mcp_tools import (
_bridge_to_work_order,
_parse_command,
close_mcp_sessions,
create_filesystem_mcp_tools,
create_gitea_issue_via_mcp,
create_gitea_mcp_tools,
)
# ---------------------------------------------------------------------------
# _parse_command
# ---------------------------------------------------------------------------
def test_parse_command_splits_correctly():
"""_parse_command splits a command string into executable and args."""
with patch("timmy.mcp_tools.shutil.which", return_value="/usr/local/bin/gitea-mcp"):
exe, args = _parse_command("gitea-mcp -t stdio")
assert exe == "/usr/local/bin/gitea-mcp"
assert args == ["-t", "stdio"]
def test_parse_command_expands_tilde():
"""_parse_command expands ~/."""
with patch("timmy.mcp_tools.shutil.which", return_value=None):
exe, args = _parse_command("~/go/bin/gitea-mcp -t stdio")
assert "/go/bin/gitea-mcp" in exe
assert "~" not in exe
assert args == ["-t", "stdio"]
def test_parse_command_preserves_absolute_path():
"""_parse_command preserves an absolute path without calling which."""
exe, args = _parse_command("/usr/local/bin/gitea-mcp -t stdio")
assert exe == "/usr/local/bin/gitea-mcp"
assert args == ["-t", "stdio"]
# ---------------------------------------------------------------------------
# create_gitea_mcp_tools
# ---------------------------------------------------------------------------
@@ -36,24 +66,24 @@ def test_gitea_mcp_returns_none_when_no_token():
def test_gitea_mcp_returns_tools_when_configured():
"""Gitea MCP factory returns an MCPTools instance when properly configured."""
"""Gitea MCP factory returns an MCPTools instance using server_params."""
mock_mcp = MagicMock()
mock_params = MagicMock()
with (
patch("timmy.mcp_tools.settings") as mock_settings,
patch("agno.tools.mcp.MCPTools", return_value=mock_mcp) as mock_cls,
patch("timmy.mcp_tools._gitea_server_params", return_value=mock_params),
):
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.mcp_timeout = 15
result = create_gitea_mcp_tools()
assert result is mock_mcp
mock_cls.assert_called_once()
call_kwargs = mock_cls.call_args[1]
assert call_kwargs["command"] == "gitea-mcp -t stdio"
assert call_kwargs["env"]["GITEA_ACCESS_TOKEN"] == "tok123"
assert call_kwargs["server_params"] is mock_params
assert "command" not in call_kwargs
assert "issue_write" in call_kwargs["include_tools"]
assert "pull_request_write" in call_kwargs["include_tools"]
@@ -77,11 +107,14 @@ def test_gitea_mcp_graceful_on_import_error():
def test_filesystem_mcp_returns_tools():
"""Filesystem MCP factory returns an MCPTools instance."""
"""Filesystem MCP factory returns an MCPTools instance using server_params."""
mock_mcp = MagicMock()
mock_params_cls = MagicMock()
with (
patch("timmy.mcp_tools.settings") as mock_settings,
patch("agno.tools.mcp.MCPTools", return_value=mock_mcp) as mock_cls,
patch("mcp.client.stdio.StdioServerParameters", mock_params_cls),
patch("timmy.mcp_tools.shutil.which", return_value="/usr/local/bin/npx"),
):
mock_settings.mcp_filesystem_command = "npx -y @modelcontextprotocol/server-filesystem"
mock_settings.repo_root = "/home/user/project"
@@ -90,8 +123,11 @@ def test_filesystem_mcp_returns_tools():
assert result is mock_mcp
call_kwargs = mock_cls.call_args[1]
assert "/home/user/project" in call_kwargs["command"]
assert "server_params" in call_kwargs
assert "read_file" in call_kwargs["include_tools"]
# Verify StdioServerParameters was called with repo_root as an arg
params_kwargs = mock_params_cls.call_args[1]
assert "/home/user/project" in params_kwargs["args"]
# ---------------------------------------------------------------------------
@@ -111,23 +147,29 @@ async def test_issue_via_mcp_returns_not_configured():
@pytest.mark.asyncio
async def test_issue_via_mcp_calls_tool():
"""Issue creation calls the MCP tool with correct arguments."""
"""Issue creation calls session.call_tool with correct arguments."""
import timmy.mcp_tools as mcp_mod
mock_session = MagicMock()
mock_session.connect = AsyncMock()
mock_session.call_tool = AsyncMock(return_value="Issue #42 created")
mock_session._connected = False
# Mock the inner MCP session (tools.session.call_tool)
mock_inner_session = MagicMock()
mock_inner_session.call_tool = AsyncMock(return_value="Issue #42 created")
mock_tools = MagicMock()
mock_tools.connect = AsyncMock()
mock_tools.session = mock_inner_session
mock_tools._connected = False
mock_params = MagicMock()
with (
patch("timmy.mcp_tools.settings") as mock_settings,
patch("agno.tools.mcp.MCPTools", return_value=mock_session),
patch("agno.tools.mcp.MCPTools", return_value=mock_tools),
patch("timmy.mcp_tools._gitea_server_params", return_value=mock_params),
):
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_repo = "owner/repo"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
mock_settings.mcp_timeout = 15
mock_settings.repo_root = "/tmp/test"
@@ -137,9 +179,10 @@ async def test_issue_via_mcp_calls_tool():
result = await create_gitea_issue_via_mcp("Bug title", "Bug body", "bug")
assert "Bug title" in result
mock_session.connect.assert_awaited_once()
mock_session.call_tool.assert_awaited_once()
call_args = mock_session.call_tool.call_args
mock_tools.connect.assert_awaited_once()
# Verify it calls session.call_tool (not tools.call_tool)
mock_inner_session.call_tool.assert_awaited_once()
call_args = mock_inner_session.call_tool.call_args
assert call_args[0][0] == "issue_write"
assert call_args[1]["arguments"]["method"] == "create"
assert call_args[1]["arguments"]["owner"] == "owner"
@@ -154,19 +197,21 @@ async def test_issue_via_mcp_graceful_failure():
"""Issue creation returns error string on MCP failure."""
import timmy.mcp_tools as mcp_mod
mock_session = MagicMock()
mock_session.connect = AsyncMock(side_effect=ConnectionError("no process"))
mock_session._connected = False
mock_tools = MagicMock()
mock_tools.connect = AsyncMock(side_effect=ConnectionError("no process"))
mock_tools._connected = False
mock_params = MagicMock()
with (
patch("timmy.mcp_tools.settings") as mock_settings,
patch("agno.tools.mcp.MCPTools", return_value=mock_session),
patch("agno.tools.mcp.MCPTools", return_value=mock_tools),
patch("timmy.mcp_tools._gitea_server_params", return_value=mock_params),
):
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_repo = "owner/repo"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
mock_settings.mcp_timeout = 15
mock_settings.repo_root = "/tmp/test"
@@ -209,16 +254,16 @@ def test_bridge_to_work_order(tmp_path):
@pytest.mark.asyncio
async def test_close_mcp_sessions():
"""close_mcp_sessions disconnects the cached session."""
"""close_mcp_sessions closes the cached session."""
import timmy.mcp_tools as mcp_mod
mock_session = MagicMock()
mock_session.disconnect = AsyncMock()
mock_session.close = AsyncMock()
mcp_mod._issue_session = mock_session
await close_mcp_sessions()
mock_session.disconnect.assert_awaited_once()
mock_session.close.assert_awaited_once()
assert mcp_mod._issue_session is None