feat: Add Gitea MCP integration and project cleanup
## New Features - **Gitea MCP Tools** (zero API cost): - gitea_read_file: Read files from homelab repo - gitea_list_files: Browse directories - gitea_search_code: Search by filename - gitea_get_tree: Get directory tree - **Gitea Client** (gitea_tools/client.py): REST API wrapper with OAuth - **Proxmox SSH Scripts** (scripts/): Homelab data collection utilities - **Obsidian MCP Support** (obsidian_mcp.py): Advanced vault operations - **Voice Integration Plan** (JARVIS_VOICE_INTEGRATION_PLAN.md) ## Improvements - **Increased timeout**: 5min → 10min for complex tasks (llm_interface.py) - **Removed Direct API fallback**: Gitea tools are MCP-only (zero cost) - **Updated .env.example**: Added Obsidian MCP configuration - **Enhanced .gitignore**: Protect personal memory files (SOUL.md, MEMORY.md) ## Cleanup - Deleted 24 obsolete files (temp/test/experimental scripts, outdated docs) - Untracked personal memory files (SOUL.md, MEMORY.md now in .gitignore) - Removed: AGENT_SDK_IMPLEMENTATION.md, HYBRID_SEARCH_SUMMARY.md, IMPLEMENTATION_SUMMARY.md, MIGRATION.md, test_agent_sdk.py, etc. ## Configuration - Added config/gitea_config.example.yaml (Gitea setup template) - Added config/obsidian_mcp.example.yaml (Obsidian MCP template) - Updated scheduled_tasks.yaml with new task examples Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
771
mcp_tools.py
771
mcp_tools.py
@@ -1033,22 +1033,789 @@ async def search_vault_tool(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
# Create the MCP server with all tools (file/system + web + zettelkasten)
|
||||
# ============================================
|
||||
# Google and Weather Tools (MCP Migration)
|
||||
# ============================================
|
||||
# Lazy-loaded Google clients
|
||||
_gmail_client: Optional[Any] = None
|
||||
_calendar_client: Optional[Any] = None
|
||||
_people_client: Optional[Any] = None
|
||||
|
||||
|
||||
def _initialize_google_clients():
|
||||
"""Lazy-load Google API clients when needed."""
|
||||
global _gmail_client, _calendar_client, _people_client
|
||||
|
||||
if _gmail_client is not None:
|
||||
return _gmail_client, _calendar_client, _people_client
|
||||
|
||||
try:
|
||||
from google_tools.gmail_client import GmailClient
|
||||
from google_tools.calendar_client import CalendarClient
|
||||
from google_tools.people_client import PeopleClient
|
||||
from google_tools.oauth_manager import GoogleOAuthManager
|
||||
|
||||
oauth_manager = GoogleOAuthManager()
|
||||
credentials = oauth_manager.get_credentials()
|
||||
|
||||
if not credentials:
|
||||
return None, None, None
|
||||
|
||||
_gmail_client = GmailClient(oauth_manager)
|
||||
_calendar_client = CalendarClient(oauth_manager)
|
||||
_people_client = PeopleClient(oauth_manager)
|
||||
|
||||
return _gmail_client, _calendar_client, _people_client
|
||||
except Exception as e:
|
||||
print(f"[MCP Google] Failed to initialize: {e}")
|
||||
return None, None, None
|
||||
|
||||
|
||||
@tool(
|
||||
name="get_weather",
|
||||
description="Get current weather for a location using OpenWeatherMap API. Returns temperature, conditions, and description.",
|
||||
input_schema={"location": str},
|
||||
)
|
||||
async def get_weather(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get current weather for a location using OpenWeatherMap API."""
|
||||
location = args.get("location", "Phoenix, US")
|
||||
import os
|
||||
import requests
|
||||
|
||||
api_key = os.getenv("OPENWEATHERMAP_API_KEY")
|
||||
if not api_key:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": "Error: OPENWEATHERMAP_API_KEY not found in environment variables"
|
||||
}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
try:
|
||||
base_url = "http://api.openweathermap.org/data/2.5/weather"
|
||||
params = {
|
||||
"q": location,
|
||||
"appid": api_key,
|
||||
"units": "imperial"
|
||||
}
|
||||
|
||||
response = requests.get(base_url, params=params, timeout=10)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
temp = data["main"]["temp"]
|
||||
feels_like = data["main"]["feels_like"]
|
||||
humidity = data["main"]["humidity"]
|
||||
conditions = data["weather"][0]["main"]
|
||||
description = data["weather"][0]["description"]
|
||||
city_name = data["name"]
|
||||
|
||||
summary = (
|
||||
f"Weather in {city_name}:\n"
|
||||
f"Temperature: {temp}°F (feels like {feels_like}°F)\n"
|
||||
f"Conditions: {conditions} - {description}\n"
|
||||
f"Humidity: {humidity}%"
|
||||
)
|
||||
|
||||
return {
|
||||
"content": [{"type": "text", "text": summary}]
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": f"Error getting weather: {str(e)}"
|
||||
}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
|
||||
@tool(
|
||||
name="send_email",
|
||||
description="Send an email via Gmail API. Requires prior OAuth setup (--setup-google).",
|
||||
input_schema={"to": str, "subject": str, "body": str, "cc": str, "reply_to_message_id": str},
|
||||
)
|
||||
async def send_email(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Send an email via Gmail API."""
|
||||
to = args["to"]
|
||||
subject = args["subject"]
|
||||
body = args["body"]
|
||||
cc = args.get("cc")
|
||||
reply_to_message_id = args.get("reply_to_message_id")
|
||||
|
||||
gmail_client, _, _ = _initialize_google_clients()
|
||||
|
||||
if not gmail_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = gmail_client.send_email(
|
||||
to=to, subject=subject, body=body, cc=cc,
|
||||
reply_to_message_id=reply_to_message_id,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
msg_id = result.get("message_id", "unknown")
|
||||
text = f"Email sent successfully to {to}\nMessage ID: {msg_id}\nSubject: {subject}"
|
||||
else:
|
||||
text = f"Error sending email: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="read_emails",
|
||||
description="Search and read emails from Gmail using Gmail query syntax (e.g., 'from:user@example.com after:2026/02/10').",
|
||||
input_schema={"query": str, "max_results": int, "include_body": bool},
|
||||
)
|
||||
async def read_emails(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Search and read emails from Gmail."""
|
||||
query = args.get("query", "")
|
||||
max_results = args.get("max_results", 10)
|
||||
include_body = args.get("include_body", False)
|
||||
|
||||
gmail_client, _, _ = _initialize_google_clients()
|
||||
|
||||
if not gmail_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = gmail_client.search_emails(query=query, max_results=max_results, include_body=include_body)
|
||||
|
||||
if result["success"]:
|
||||
summary = result.get("summary", "")
|
||||
if len(summary) > _MAX_TOOL_OUTPUT:
|
||||
summary = summary[:_MAX_TOOL_OUTPUT] + "\n... (results truncated)"
|
||||
else:
|
||||
summary = f"Error reading emails: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": summary}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="get_email",
|
||||
description="Get full content of a specific email by its Gmail message ID.",
|
||||
input_schema={"message_id": str, "format_type": str},
|
||||
)
|
||||
async def get_email(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get full content of a specific email."""
|
||||
message_id = args["message_id"]
|
||||
format_type = args.get("format_type", "text")
|
||||
|
||||
gmail_client, _, _ = _initialize_google_clients()
|
||||
|
||||
if not gmail_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = gmail_client.get_email(message_id=message_id, format_type=format_type)
|
||||
|
||||
if result["success"]:
|
||||
email_data = result.get("email", {})
|
||||
text = (
|
||||
f"From: {email_data.get('from', 'Unknown')}\n"
|
||||
f"To: {email_data.get('to', 'Unknown')}\n"
|
||||
f"Subject: {email_data.get('subject', 'No subject')}\n"
|
||||
f"Date: {email_data.get('date', 'Unknown')}\n\n"
|
||||
f"{email_data.get('body', 'No content')}"
|
||||
)
|
||||
if len(text) > _MAX_TOOL_OUTPUT:
|
||||
text = text[:_MAX_TOOL_OUTPUT] + "\n... (content truncated)"
|
||||
else:
|
||||
text = f"Error getting email: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="read_calendar",
|
||||
description="Read upcoming events from Google Calendar. Shows events from today onwards.",
|
||||
input_schema={"days_ahead": int, "calendar_id": str, "max_results": int},
|
||||
)
|
||||
async def read_calendar(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Read upcoming calendar events."""
|
||||
days_ahead = args.get("days_ahead", 7)
|
||||
calendar_id = args.get("calendar_id", "primary")
|
||||
max_results = args.get("max_results", 20)
|
||||
|
||||
_, calendar_client, _ = _initialize_google_clients()
|
||||
|
||||
if not calendar_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = calendar_client.list_events(
|
||||
days_ahead=days_ahead, calendar_id=calendar_id, max_results=max_results,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
summary = result.get("summary", "No events found")
|
||||
if len(summary) > _MAX_TOOL_OUTPUT:
|
||||
summary = summary[:_MAX_TOOL_OUTPUT] + "\n... (results truncated)"
|
||||
text = f"Upcoming events (next {days_ahead} days):\n\n{summary}"
|
||||
else:
|
||||
text = f"Error reading calendar: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="create_calendar_event",
|
||||
description="Create a new event in Google Calendar. Use ISO 8601 format for times.",
|
||||
input_schema={
|
||||
"summary": str, "start_time": str, "end_time": str,
|
||||
"description": str, "location": str, "calendar_id": str,
|
||||
},
|
||||
)
|
||||
async def create_calendar_event(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Create a new calendar event."""
|
||||
summary = args["summary"]
|
||||
start_time = args["start_time"]
|
||||
end_time = args["end_time"]
|
||||
description = args.get("description", "")
|
||||
location = args.get("location", "")
|
||||
calendar_id = args.get("calendar_id", "primary")
|
||||
|
||||
_, calendar_client, _ = _initialize_google_clients()
|
||||
|
||||
if not calendar_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = calendar_client.create_event(
|
||||
summary=summary, start_time=start_time, end_time=end_time,
|
||||
description=description, location=location, calendar_id=calendar_id,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
event_id = result.get("event_id", "unknown")
|
||||
html_link = result.get("html_link", "")
|
||||
start = result.get("start", start_time)
|
||||
text = (
|
||||
f"Calendar event created successfully!\n"
|
||||
f"Title: {summary}\nStart: {start}\n"
|
||||
f"Event ID: {event_id}\nLink: {html_link}"
|
||||
)
|
||||
else:
|
||||
text = f"Error creating calendar event: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="search_calendar",
|
||||
description="Search calendar events by text query. Searches event titles and descriptions.",
|
||||
input_schema={"query": str, "calendar_id": str},
|
||||
)
|
||||
async def search_calendar(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Search calendar events by text query."""
|
||||
query = args["query"]
|
||||
calendar_id = args.get("calendar_id", "primary")
|
||||
|
||||
_, calendar_client, _ = _initialize_google_clients()
|
||||
|
||||
if not calendar_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = calendar_client.search_events(query=query, calendar_id=calendar_id)
|
||||
|
||||
if result["success"]:
|
||||
summary = result.get("summary", "No events found")
|
||||
if len(summary) > _MAX_TOOL_OUTPUT:
|
||||
summary = summary[:_MAX_TOOL_OUTPUT] + "\n... (results truncated)"
|
||||
text = f"Calendar search results for '{query}':\n\n{summary}"
|
||||
else:
|
||||
text = f"Error searching calendar: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
# ============================================
|
||||
# Contacts Tools (MCP)
|
||||
# ============================================
|
||||
|
||||
|
||||
@tool(
|
||||
name="create_contact",
|
||||
description="Create a new Google contact. Requires prior OAuth setup (--setup-google).",
|
||||
input_schema={"given_name": str, "family_name": str, "email": str, "phone": str, "notes": str},
|
||||
)
|
||||
async def create_contact(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Create a new Google contact."""
|
||||
given_name = args["given_name"]
|
||||
family_name = args.get("family_name", "")
|
||||
email = args.get("email", "")
|
||||
phone = args.get("phone")
|
||||
notes = args.get("notes")
|
||||
|
||||
_, _, people_client = _initialize_google_clients()
|
||||
|
||||
if not people_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = people_client.create_contact(
|
||||
given_name=given_name, family_name=family_name,
|
||||
email=email, phone=phone, notes=notes,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
name = result.get("name", given_name)
|
||||
resource = result.get("resource_name", "")
|
||||
text = f"Contact created: {name}\nResource: {resource}"
|
||||
else:
|
||||
text = f"Error creating contact: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="list_contacts",
|
||||
description="List or search Google contacts. Without a query, lists all contacts sorted by last name.",
|
||||
input_schema={"max_results": int, "query": str},
|
||||
)
|
||||
async def list_contacts(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""List or search Google contacts."""
|
||||
max_results = args.get("max_results", 100)
|
||||
query = args.get("query")
|
||||
|
||||
_, _, people_client = _initialize_google_clients()
|
||||
|
||||
if not people_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = people_client.list_contacts(max_results=max_results, query=query)
|
||||
|
||||
if result["success"]:
|
||||
summary = result.get("summary", "No contacts found.")
|
||||
if len(summary) > _MAX_TOOL_OUTPUT:
|
||||
summary = summary[:_MAX_TOOL_OUTPUT] + "\n... (results truncated)"
|
||||
text = f"Contacts ({result.get('count', 0)} found):\n\n{summary}"
|
||||
else:
|
||||
text = f"Error listing contacts: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
@tool(
|
||||
name="get_contact",
|
||||
description="Get full details of a specific Google contact by resource name.",
|
||||
input_schema={"resource_name": str},
|
||||
)
|
||||
async def get_contact(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get full details of a specific Google contact."""
|
||||
resource_name = args["resource_name"]
|
||||
|
||||
_, _, people_client = _initialize_google_clients()
|
||||
|
||||
if not people_client:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: Google not authorized. Run: python bot_runner.py --setup-google"}],
|
||||
"isError": True
|
||||
}
|
||||
|
||||
result = people_client.get_contact(resource_name=resource_name)
|
||||
|
||||
if result["success"]:
|
||||
c = result.get("contact", {})
|
||||
output = []
|
||||
name = c.get("display_name") or f"{c.get('given_name', '')} {c.get('family_name', '')}".strip()
|
||||
output.append(f"Name: {name or '(no name)'}")
|
||||
if c.get("email"):
|
||||
output.append(f"Email: {c['email']}")
|
||||
if c.get("phone"):
|
||||
output.append(f"Phone: {c['phone']}")
|
||||
if c.get("notes"):
|
||||
output.append(f"Notes: {c['notes']}")
|
||||
output.append(f"Resource: {c.get('resource_name', resource_name)}")
|
||||
text = "\n".join(output)
|
||||
else:
|
||||
text = f"Error getting contact: {result.get('error', 'Unknown error')}"
|
||||
|
||||
return {"content": [{"type": "text", "text": text}], "isError": not result["success"]}
|
||||
|
||||
|
||||
# ============================================
|
||||
# Gitea Tools (MCP) - Private repo access
|
||||
# ============================================
|
||||
|
||||
# Lazy-loaded Gitea client
|
||||
_gitea_client: Optional[Any] = None
|
||||
|
||||
|
||||
def _get_gitea_client():
|
||||
"""Lazy-load Gitea client when first needed."""
|
||||
global _gitea_client
|
||||
|
||||
if _gitea_client is not None:
|
||||
return _gitea_client
|
||||
|
||||
try:
|
||||
from gitea_tools.client import get_gitea_client
|
||||
_gitea_client = get_gitea_client()
|
||||
return _gitea_client
|
||||
except Exception as e:
|
||||
print(f"[MCP Gitea] Failed to initialize: {e}")
|
||||
return None
|
||||
|
||||
|
||||
@tool(
|
||||
name="gitea_read_file",
|
||||
description="Read a file from a Gitea repository. Use this to access files from Jordan's homelab repo or any configured Gitea repo. Returns the file content as text.",
|
||||
input_schema={
|
||||
"file_path": str,
|
||||
"repo": str,
|
||||
"branch": str,
|
||||
},
|
||||
)
|
||||
async def gitea_read_file_tool(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Read a file from a Gitea repository.
|
||||
|
||||
Zero-cost MCP tool for accessing private Gitea repos.
|
||||
"""
|
||||
file_path = args.get("file_path", "")
|
||||
repo = args.get("repo")
|
||||
branch = args.get("branch")
|
||||
|
||||
if not file_path:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: file_path is required"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
client = _get_gitea_client()
|
||||
if not client:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": (
|
||||
"Error: Gitea not configured. "
|
||||
"Copy config/gitea_config.example.yaml to config/gitea_config.yaml "
|
||||
"and add your Personal Access Token."
|
||||
),
|
||||
}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
# Parse owner/repo if provided
|
||||
owner = None
|
||||
if repo and "/" in repo:
|
||||
parts = repo.split("/", 1)
|
||||
owner = parts[0]
|
||||
repo = parts[1]
|
||||
|
||||
result = await client.get_file_content(
|
||||
file_path=file_path,
|
||||
owner=owner,
|
||||
repo=repo,
|
||||
branch=branch,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
content = result["content"]
|
||||
metadata = result.get("metadata", {})
|
||||
path_info = metadata.get("path", file_path)
|
||||
size = metadata.get("size", 0)
|
||||
|
||||
header = f"File: {path_info} ({size:,} bytes)"
|
||||
if metadata.get("truncated"):
|
||||
header += " [TRUNCATED]"
|
||||
|
||||
return {
|
||||
"content": [{"type": "text", "text": f"{header}\n\n{content}"}],
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"content": [{"type": "text", "text": f"Error: {result['error']}"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
|
||||
@tool(
|
||||
name="gitea_list_files",
|
||||
description="List files and folders in a directory in a Gitea repository. Use this to explore the structure of Jordan's homelab repo or any configured Gitea repo.",
|
||||
input_schema={
|
||||
"path": str,
|
||||
"repo": str,
|
||||
"branch": str,
|
||||
},
|
||||
)
|
||||
async def gitea_list_files_tool(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""List files and directories in a Gitea repo path.
|
||||
|
||||
Zero-cost MCP tool for browsing private Gitea repos.
|
||||
"""
|
||||
path = args.get("path", "")
|
||||
repo = args.get("repo")
|
||||
branch = args.get("branch")
|
||||
|
||||
client = _get_gitea_client()
|
||||
if not client:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": (
|
||||
"Error: Gitea not configured. "
|
||||
"Copy config/gitea_config.example.yaml to config/gitea_config.yaml "
|
||||
"and add your Personal Access Token."
|
||||
),
|
||||
}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
# Parse owner/repo if provided
|
||||
owner = None
|
||||
if repo and "/" in repo:
|
||||
parts = repo.split("/", 1)
|
||||
owner = parts[0]
|
||||
repo = parts[1]
|
||||
|
||||
result = await client.list_files(
|
||||
path=path,
|
||||
owner=owner,
|
||||
repo=repo,
|
||||
branch=branch,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
files = result["files"]
|
||||
repo_name = result.get("repo", "")
|
||||
display_path = result.get("path", "/")
|
||||
count = result.get("count", 0)
|
||||
|
||||
# Format output
|
||||
lines = [f"Directory: {repo_name}/{display_path} ({count} items)\n"]
|
||||
for f in files:
|
||||
if f["type"] == "dir":
|
||||
lines.append(f" DIR {f['name']}/")
|
||||
else:
|
||||
size_str = f"({f['size']:,} bytes)" if f["size"] else ""
|
||||
lines.append(f" FILE {f['name']} {size_str}")
|
||||
|
||||
return {
|
||||
"content": [{"type": "text", "text": "\n".join(lines)}],
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"content": [{"type": "text", "text": f"Error: {result['error']}"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
|
||||
@tool(
|
||||
name="gitea_search_code",
|
||||
description="Search for files by name/path in a Gitea repository. Searches file and directory names. For content search, use gitea_read_file on specific files.",
|
||||
input_schema={
|
||||
"query": str,
|
||||
"repo": str,
|
||||
},
|
||||
)
|
||||
async def gitea_search_code_tool(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Search for code/files in a Gitea repository.
|
||||
|
||||
Zero-cost MCP tool. Searches file/directory names in the repo tree.
|
||||
"""
|
||||
query = args.get("query", "")
|
||||
repo = args.get("repo")
|
||||
|
||||
if not query:
|
||||
return {
|
||||
"content": [{"type": "text", "text": "Error: query is required"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
client = _get_gitea_client()
|
||||
if not client:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": (
|
||||
"Error: Gitea not configured. "
|
||||
"Copy config/gitea_config.example.yaml to config/gitea_config.yaml "
|
||||
"and add your Personal Access Token."
|
||||
),
|
||||
}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
# Parse owner/repo if provided
|
||||
owner = None
|
||||
if repo and "/" in repo:
|
||||
parts = repo.split("/", 1)
|
||||
owner = parts[0]
|
||||
repo = parts[1]
|
||||
|
||||
result = await client.search_code(
|
||||
query=query,
|
||||
owner=owner,
|
||||
repo=repo,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
results = result.get("results", [])
|
||||
count = result.get("count", 0)
|
||||
repo_name = result.get("repo", "")
|
||||
|
||||
if not results:
|
||||
message = result.get("message", f"No results for '{query}'")
|
||||
return {
|
||||
"content": [{"type": "text", "text": message}],
|
||||
}
|
||||
|
||||
lines = [f"Search results for '{query}' in {repo_name} ({count} matches):\n"]
|
||||
for r in results:
|
||||
type_icon = "DIR " if r["type"] == "dir" else "FILE"
|
||||
size_str = f"({r['size']:,} bytes)" if r.get("size") else ""
|
||||
lines.append(f" {type_icon} {r['path']} {size_str}")
|
||||
|
||||
return {
|
||||
"content": [{"type": "text", "text": "\n".join(lines)}],
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"content": [{"type": "text", "text": f"Error: {result['error']}"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
|
||||
@tool(
|
||||
name="gitea_get_tree",
|
||||
description="Get the directory tree structure from a Gitea repository. Shows all files and folders. Use recursive=true for the full tree.",
|
||||
input_schema={
|
||||
"repo": str,
|
||||
"branch": str,
|
||||
"recursive": bool,
|
||||
},
|
||||
)
|
||||
async def gitea_get_tree_tool(args: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get directory tree from a Gitea repository.
|
||||
|
||||
Zero-cost MCP tool for viewing repo structure.
|
||||
"""
|
||||
repo = args.get("repo")
|
||||
branch = args.get("branch")
|
||||
recursive = args.get("recursive", False)
|
||||
|
||||
client = _get_gitea_client()
|
||||
if not client:
|
||||
return {
|
||||
"content": [{
|
||||
"type": "text",
|
||||
"text": (
|
||||
"Error: Gitea not configured. "
|
||||
"Copy config/gitea_config.example.yaml to config/gitea_config.yaml "
|
||||
"and add your Personal Access Token."
|
||||
),
|
||||
}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
# Parse owner/repo if provided
|
||||
owner = None
|
||||
if repo and "/" in repo:
|
||||
parts = repo.split("/", 1)
|
||||
owner = parts[0]
|
||||
repo = parts[1]
|
||||
|
||||
result = await client.get_tree(
|
||||
owner=owner,
|
||||
repo=repo,
|
||||
branch=branch,
|
||||
recursive=recursive,
|
||||
)
|
||||
|
||||
if result["success"]:
|
||||
entries = result.get("entries", [])
|
||||
repo_name = result.get("repo", "")
|
||||
branch_name = result.get("branch", "main")
|
||||
total = result.get("total", 0)
|
||||
truncated = result.get("truncated", False)
|
||||
|
||||
lines = [f"Tree: {repo_name} (branch: {branch_name}, {total} entries)"]
|
||||
if truncated:
|
||||
lines[0] += " [TRUNCATED - tree too large]"
|
||||
lines.append("")
|
||||
|
||||
for entry in entries:
|
||||
if entry["type"] == "dir":
|
||||
lines.append(f" {entry['path']}/")
|
||||
else:
|
||||
size_str = f"({entry['size']:,} bytes)" if entry.get("size") else ""
|
||||
lines.append(f" {entry['path']} {size_str}")
|
||||
|
||||
# Truncate output if too long
|
||||
text = "\n".join(lines)
|
||||
if len(text) > _MAX_TOOL_OUTPUT:
|
||||
text = text[:_MAX_TOOL_OUTPUT] + "\n\n... (tree truncated, use gitea_list_files for specific directories)"
|
||||
|
||||
return {
|
||||
"content": [{"type": "text", "text": text}],
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"content": [{"type": "text", "text": f"Error: {result['error']}"}],
|
||||
"isError": True,
|
||||
}
|
||||
|
||||
|
||||
# Create the MCP server with all tools
|
||||
file_system_server = create_sdk_mcp_server(
|
||||
name="file_system",
|
||||
version="1.4.0",
|
||||
version="2.0.0",
|
||||
tools=[
|
||||
# File and system tools
|
||||
read_file_tool,
|
||||
write_file_tool,
|
||||
edit_file_tool,
|
||||
list_directory_tool,
|
||||
run_command_tool,
|
||||
# Web tool
|
||||
web_fetch_tool,
|
||||
# Zettelkasten tools
|
||||
fleeting_note_tool,
|
||||
daily_note_tool,
|
||||
literature_note_tool,
|
||||
permanent_note_tool,
|
||||
search_vault_tool,
|
||||
search_by_tags_tool,
|
||||
# Weather
|
||||
get_weather,
|
||||
# Gmail tools
|
||||
send_email,
|
||||
read_emails,
|
||||
get_email,
|
||||
# Calendar tools
|
||||
read_calendar,
|
||||
create_calendar_event,
|
||||
search_calendar,
|
||||
# Contacts tools
|
||||
create_contact,
|
||||
list_contacts,
|
||||
get_contact,
|
||||
# Gitea tools
|
||||
gitea_read_file_tool,
|
||||
gitea_list_files_tool,
|
||||
gitea_search_code_tool,
|
||||
gitea_get_tree_tool,
|
||||
]
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user