66 lines
2.6 KiB
Python
66 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Git/Gitea MCP Server — Browse repos via Gitea API."""
|
|
from fastmcp import FastMCP
|
|
import httpx, os, base64
|
|
|
|
GITEA_URL = os.getenv('GITEA_URL', 'https://git.itpropartner.com')
|
|
API = f'{GITEA_URL}/api/v1'
|
|
|
|
mcp = FastMCP('Git/Gitea', version='1.0.0')
|
|
|
|
async def _get(path: str):
|
|
async with httpx.AsyncClient(verify=False, timeout=15) as c:
|
|
r = await c.get(f'{API}{path}')
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
@mcp.tool(description='List all public repos on git.itpropartner.com')
|
|
async def list_repos() -> str:
|
|
"""List all Gitea repos."""
|
|
repos = await _get('/repos/search?limit=50&q=&template=false')
|
|
if isinstance(repos, dict) and 'data' in repos:
|
|
repos = repos['data']
|
|
if not repos:
|
|
return 'No repos found. Gitea may need initial setup at https://git.itpropartner.com'
|
|
lines = [f'{len(repos)} repos:']
|
|
for r in repos[:30]:
|
|
lines.append(f' {r["full_name"]} — {r.get("description","")} ({r.get("stars_count",0)}★)')
|
|
return '\n'.join(lines)
|
|
|
|
@mcp.tool(description='Search Gitea repos by name or description')
|
|
async def search_repos(query: str) -> str:
|
|
"""Search repos."""
|
|
repos = await _get(f'/repos/search?q={query}&limit=20')
|
|
if isinstance(repos, dict) and 'data' in repos:
|
|
repos = repos['data']
|
|
if not repos:
|
|
return f'No repos matching "{query}"'
|
|
lines = [f'Results for "{query}":']
|
|
for r in repos[:20]:
|
|
lines.append(f' {r["full_name"]} — {r.get("description","")}')
|
|
return '\n'.join(lines)
|
|
|
|
@mcp.tool(description='Browse files and folders in a Gitea repo')
|
|
async def browse_repo(owner: str, repo: str, path: str = '') -> str:
|
|
"""List contents of a repo directory."""
|
|
contents = await _get(f'/repos/{owner}/{repo}/contents/{path}')
|
|
if not contents:
|
|
return f'Empty: {owner}/{repo}/{path}'
|
|
lines = [f'{owner}/{repo}/{path or "/"}:']
|
|
for item in contents[:50]:
|
|
typ = '[DIR]' if item['type'] == 'dir' else f'[{item["type"]}]'
|
|
lines.append(f' {typ} {item["name"]} ({item.get("size",0):,} bytes)')
|
|
return '\n'.join(lines)
|
|
|
|
@mcp.tool(description='Read a file from a Gitea repo')
|
|
async def read_file(owner: str, repo: str, path: str) -> str:
|
|
"""Read file contents."""
|
|
data = await _get(f'/repos/{owner}/{repo}/contents/{path}')
|
|
if 'content' in data:
|
|
content = base64.b64decode(data['content']).decode('utf-8', errors='replace')
|
|
return f'{owner}/{repo}/{path} ({data["size"]:,} bytes):\n\n{content[:10000]}'
|
|
return f'Could not read {owner}/{repo}/{path}'
|
|
|
|
if __name__ == '__main__':
|
|
mcp.run(transport='http', host='0.0.0.0', port=8903, path='/mcp')
|