import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import express from "express"; import crypto from "crypto"; import { CallToolRequestSchema, ListToolsRequestSchema, ListResourcesRequestSchema, ReadResourceRequestSchema, SubscribeRequestSchema, UnsubscribeRequestSchema, Tool, } from "@modelcontextprotocol/sdk/types.js"; import axios from "axios"; const GITEA_HOST = process.env.GITEA_HOST || "https://git.infra.mintel.me"; const GITEA_ACCESS_TOKEN = process.env.GITEA_ACCESS_TOKEN || process.env.GITEA_TOKEN; if (!GITEA_ACCESS_TOKEN) { console.error( "Warning: Neither GITEA_ACCESS_TOKEN nor GITEA_TOKEN environment variable is set. Pipeline tools will return unauthorized errors.", ); } const giteaClient = axios.create({ baseURL: `${GITEA_HOST.replace(/\/$/, "")}/api/v1`, headers: { Authorization: `token ${GITEA_ACCESS_TOKEN}`, }, }); const LIST_PIPELINES_TOOL: Tool = { name: "gitea_list_pipelines", description: "List recent action runs (pipelines) for a specific repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner (e.g., 'mmintel')", }, repo: { type: "string", description: "Repository name (e.g., 'at-mintel')", }, limit: { type: "number", description: "Number of runs to fetch (default: 5)", }, branch: { type: "string", description: "Optional: Filter by branch name (e.g., 'main')", }, event: { type: "string", description: "Optional: Filter by trigger event (e.g., 'push', 'pull_request')", }, }, required: ["owner", "repo"], }, }; const GET_PIPELINE_LOGS_TOOL: Tool = { name: "gitea_get_pipeline_logs", description: "Get detailed logs for a specific pipeline run or job", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, run_id: { type: "number", description: "ID of the action run" }, }, required: ["owner", "repo", "run_id"], }, }; const WAIT_PIPELINE_COMPLETION_TOOL: Tool = { name: "gitea_wait_pipeline_completion", description: "BLOCKS and waits until a pipeline run completes, fails, or is cancelled. Use this instead of polling manually to save tokens.", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, run_id: { type: "number", description: "ID of the action run" }, timeout_minutes: { type: "number", description: "Maximum time to wait before aborting (default: 10)", }, }, required: ["owner", "repo", "run_id"], }, }; const LIST_ISSUES_TOOL: Tool = { name: "gitea_list_issues", description: "List issues for a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, state: { type: "string", description: "Filter by state: open, closed, or all (default: open)", }, limit: { type: "number", description: "Number of issues to fetch (default: 10)", }, }, required: ["owner", "repo"], }, }; const CREATE_ISSUE_TOOL: Tool = { name: "gitea_create_issue", description: "Create a new issue in a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, title: { type: "string", description: "Issue title" }, body: { type: "string", description: "Issue description/body" }, }, required: ["owner", "repo", "title"], }, }; const GET_FILE_CONTENT_TOOL: Tool = { name: "gitea_get_file_content", description: "Get the raw content of a file from a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, filepath: { type: "string", description: "Path to the file in the repository", }, ref: { type: "string", description: "The name of the commit/branch/tag (default: main)", }, }, required: ["owner", "repo", "filepath"], }, }; const UPDATE_ISSUE_TOOL: Tool = { name: "gitea_update_issue", description: "Update an existing issue (e.g. change state, title, or body)", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, index: { type: "number", description: "Issue index/number" }, state: { type: "string", description: "Optional: 'open' or 'closed'" }, title: { type: "string", description: "Optional: New title" }, body: { type: "string", description: "Optional: New body text" }, }, required: ["owner", "repo", "index"], }, }; const CREATE_ISSUE_COMMENT_TOOL: Tool = { name: "gitea_create_issue_comment", description: "Add a comment to an existing issue or pull request", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, index: { type: "number", description: "Issue or PR index/number" }, body: { type: "string", description: "Comment body text" }, }, required: ["owner", "repo", "index", "body"], }, }; const CREATE_PULL_REQUEST_TOOL: Tool = { name: "gitea_create_pull_request", description: "Create a new Pull Request", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, head: { type: "string", description: "The branch you want to merge (e.g., 'feature/my-changes')", }, base: { type: "string", description: "The branch to merge into (e.g., 'main')", }, title: { type: "string", description: "PR title" }, body: { type: "string", description: "Optional: PR description" }, }, required: ["owner", "repo", "head", "base", "title"], }, }; const SEARCH_REPOS_TOOL: Tool = { name: "gitea_search_repos", description: "Search for repositories accessible to the authenticated user", inputSchema: { type: "object", properties: { query: { type: "string", description: "Search term" }, limit: { type: "number", description: "Maximum number of results (default: 10)", }, }, required: ["query"], }, }; // --- Wiki --- const LIST_WIKI_PAGES_TOOL: Tool = { name: "gitea_list_wiki_pages", description: "List all wiki pages of a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, }, required: ["owner", "repo"], }, }; const GET_WIKI_PAGE_TOOL: Tool = { name: "gitea_get_wiki_page", description: "Get the content of a specific wiki page", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, page_name: { type: "string", description: "Name/slug of the wiki page (e.g., 'Home')", }, }, required: ["owner", "repo", "page_name"], }, }; const CREATE_WIKI_PAGE_TOOL: Tool = { name: "gitea_create_wiki_page", description: "Create a new wiki page in a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, title: { type: "string", description: "Page title" }, content: { type: "string", description: "Page content in Markdown (base64 encoded internally)", }, message: { type: "string", description: "Optional commit message" }, }, required: ["owner", "repo", "title", "content"], }, }; const EDIT_WIKI_PAGE_TOOL: Tool = { name: "gitea_edit_wiki_page", description: "Edit an existing wiki page", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, page_name: { type: "string", description: "Current name/slug of the wiki page", }, title: { type: "string", description: "Optional: new title" }, content: { type: "string", description: "New content in Markdown" }, message: { type: "string", description: "Optional commit message" }, }, required: ["owner", "repo", "page_name", "content"], }, }; // --- Packages --- const LIST_PACKAGES_TOOL: Tool = { name: "gitea_list_packages", description: "List packages published to the Gitea package registry for a user or org", inputSchema: { type: "object", properties: { owner: { type: "string", description: "User or organization name" }, type: { type: "string", description: "Optional: Package type filter (e.g., 'npm', 'docker', 'generic')", }, limit: { type: "number", description: "Number of packages to return (default: 10)", }, }, required: ["owner"], }, }; const LIST_PACKAGE_VERSIONS_TOOL: Tool = { name: "gitea_list_package_versions", description: "List all published versions of a specific package", inputSchema: { type: "object", properties: { owner: { type: "string", description: "User or organization name" }, type: { type: "string", description: "Package type (e.g., 'npm', 'docker')", }, name: { type: "string", description: "Package name" }, }, required: ["owner", "type", "name"], }, }; // --- Releases --- const LIST_RELEASES_TOOL: Tool = { name: "gitea_list_releases", description: "List releases for a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, limit: { type: "number", description: "Number of releases to fetch (default: 10)", }, }, required: ["owner", "repo"], }, }; const GET_LATEST_RELEASE_TOOL: Tool = { name: "gitea_get_latest_release", description: "Get the latest release for a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, }, required: ["owner", "repo"], }, }; const CREATE_RELEASE_TOOL: Tool = { name: "gitea_create_release", description: "Create a new release for a repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, tag_name: { type: "string", description: "Git tag to build the release from (e.g., 'v1.2.3')", }, name: { type: "string", description: "Release title" }, body: { type: "string", description: "Optional: Release notes/description in Markdown", }, draft: { type: "boolean", description: "Optional: Create as draft (default: false)", }, prerelease: { type: "boolean", description: "Optional: Mark as prerelease (default: false)", }, }, required: ["owner", "repo", "tag_name", "name"], }, }; // --- Projects --- const LIST_PROJECTS_TOOL: Tool = { name: "gitea_list_projects", description: "List projects (kanban boards) for a user, organization, or repository", inputSchema: { type: "object", properties: { owner: { type: "string", description: "User or organization name" }, repo: { type: "string", description: "Optional: Repository name (for repo-level projects)", }, type: { type: "string", description: "Optional: 'individual' or 'repository' or 'organization'", }, }, required: ["owner"], }, }; const GET_PROJECT_COLUMNS_TOOL: Tool = { name: "gitea_get_project_columns", description: "Get the columns (board columns) of a specific project", inputSchema: { type: "object", properties: { project_id: { type: "number", description: "Numeric project ID from gitea_list_projects", }, }, required: ["project_id"], }, }; // Subscription State const subscriptions = new Set(); const runStatusCache = new Map(); // uri -> status const server = new Server( { name: "gitea-mcp-native", version: "1.0.0", }, { capabilities: { tools: {}, resources: { subscribe: true }, // Enable subscriptions }, }, ); // --- Tools --- server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: [ LIST_PIPELINES_TOOL, GET_PIPELINE_LOGS_TOOL, WAIT_PIPELINE_COMPLETION_TOOL, LIST_ISSUES_TOOL, CREATE_ISSUE_TOOL, GET_FILE_CONTENT_TOOL, UPDATE_ISSUE_TOOL, CREATE_ISSUE_COMMENT_TOOL, CREATE_PULL_REQUEST_TOOL, SEARCH_REPOS_TOOL, // Wiki LIST_WIKI_PAGES_TOOL, GET_WIKI_PAGE_TOOL, CREATE_WIKI_PAGE_TOOL, EDIT_WIKI_PAGE_TOOL, // Packages LIST_PACKAGES_TOOL, LIST_PACKAGE_VERSIONS_TOOL, // Releases LIST_RELEASES_TOOL, GET_LATEST_RELEASE_TOOL, CREATE_RELEASE_TOOL, // Projects LIST_PROJECTS_TOOL, GET_PROJECT_COLUMNS_TOOL, ], }; }); server.setRequestHandler(CallToolRequestSchema, async (request) => { if (request.params.name === "gitea_list_pipelines") { const { owner, repo, limit = 5, branch, event, } = request.params.arguments as any; try { const apiParams: Record = { limit }; if (branch) apiParams.branch = branch; if (event) apiParams.event = event; const runsResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs`, { params: apiParams, }, ); const runs = (runsResponse.data.workflow_runs || []).slice(0, limit) as any[]; const enhancedRuns = await Promise.all( runs.map(async (run: any) => { try { const jobsResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run.id}/jobs`, ); const jobs = (jobsResponse.data.jobs || []) as any[]; return { id: run.id, name: run.name, status: run.status, created_at: run.created_at, jobs: jobs.map((job: any) => ({ id: job.id, name: job.name, status: job.status, conclusion: job.conclusion, })), }; } catch { return { id: run.id, name: run.name, status: run.status, created_at: run.created_at, jobs: [], }; } }), ); return { content: [ { type: "text", text: JSON.stringify(enhancedRuns, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error fetching pipelines: ${error.message}` }, ], }; } } if (request.params.name === "gitea_get_pipeline_logs") { const { owner, repo, run_id } = request.params.arguments as any; try { const jobsResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run_id}/jobs`, ); const jobs = (jobsResponse.data.jobs || []) as any[]; const logs = await Promise.all( jobs.map(async (job: any) => { let console_log = undefined; // Fetch log text for failed, running, or stuck jobs if ( job.conclusion === "failure" || job.status === "failure" || job.status === "in_progress" || job.status === "running" ) { try { const logResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/jobs/${job.id}/logs`, ); if (typeof logResponse.data === "string") { let fullLog = logResponse.data; // First, truncate to 20000 characters to prevent OOM / regex crashes on multi-megabyte strings and avoid docker exec pipe buffer limits if (fullLog.length > 20000) { fullLog = "...[truncated]...\n" + fullLog.slice(-20000); } // Strip ANSI escape codes fullLog = fullLog.replace(/\x1B\[[0-9;]*[a-zA-Z]/g, ""); // Safely strip non-printable ASCII control characters (0x00-0x1F, except 0x09 \t, 0x0A \n, 0x0D \r). // DO NOT strip characters > 0x7F, as they are part of valid UTF-8 multibyte characters (like emojis 🧹). fullLog = fullLog.replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, ""); console_log = fullLog; } } catch (err: any) { console_log = `Error fetching raw console log: ${err.message}`; } } return { job_id: job.id, job_name: job.name, status: job.status, conclusion: job.conclusion, console_log, // appended to the response steps: (job.steps || []).map((step: any) => ({ name: step.name, status: step.status, conclusion: step.conclusion, })), }; }), ); return { content: [{ type: "text", text: JSON.stringify(logs, null, 2) }], }; } catch (error: any) { return { isError: true, content: [{ type: "text", text: `Error: ${error.message}` }], }; } } if (request.params.name === "gitea_wait_pipeline_completion") { const { owner, repo, run_id, timeout_minutes = 10, } = request.params.arguments as any; const startTime = Date.now(); const timeoutMs = timeout_minutes * 60 * 1000; try { while (true) { if (Date.now() - startTime > timeoutMs) { return { content: [ { type: "text", text: `Wait timed out after ${timeout_minutes} minutes.`, }, ], }; } const response = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run_id}`, ); const status = response.data.status; const conclusion = response.data.conclusion; if (status !== "running" && status !== "waiting") { return { content: [ { type: "text", text: `Pipeline finished! Final Status: ${status}, Conclusion: ${conclusion}`, }, ], }; } // Wait 5 seconds before polling again await new Promise((resolve) => setTimeout(resolve, 5000)); } } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error checking pipeline status: ${error.message}`, }, ], }; } } if (request.params.name === "gitea_list_issues") { const { owner, repo, state = "open", limit = 10, } = request.params.arguments as any; try { const response = await giteaClient.get(`/repos/${owner}/${repo}/issues`, { params: { state, limit }, }); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [{ type: "text", text: `Error: ${error.message}` }], }; } } if (request.params.name === "gitea_create_issue") { const { owner, repo, title, body } = request.params.arguments as any; try { const response = await giteaClient.post( `/repos/${owner}/${repo}/issues`, { title, body, }, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [{ type: "text", text: `Error: ${error.message}` }], }; } } if (request.params.name === "gitea_get_file_content") { const { owner, repo, filepath, ref = "main", } = request.params.arguments as any; try { const response = await giteaClient.get( `/repos/${owner}/${repo}/contents/${filepath}`, { params: { ref }, }, ); // Gitea returns base64 encoded content for files if (response.data.type === "file" && response.data.content) { const decodedContent = Buffer.from( response.data.content, "base64", ).toString("utf-8"); return { content: [{ type: "text", text: decodedContent }] }; } return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [{ type: "text", text: `Error: ${error.message}` }], }; } } if (request.params.name === "gitea_update_issue") { const { owner, repo, index, state, title, body } = request.params .arguments as any; try { const updateData: Record = {}; if (state) updateData.state = state; if (title) updateData.title = title; if (body) updateData.body = body; // Send PATCH request to /repos/{owner}/{repo}/issues/{index} const response = await giteaClient.patch( `/repos/${owner}/${repo}/issues/${index}`, updateData, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error updating issue: ${error.message}` }, ], }; } } if (request.params.name === "gitea_create_issue_comment") { const { owner, repo, index, body } = request.params.arguments as any; try { const response = await giteaClient.post( `/repos/${owner}/${repo}/issues/${index}/comments`, { body, }, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error creating comment: ${error.message}` }, ], }; } } if (request.params.name === "gitea_create_pull_request") { const { owner, repo, head, base, title, body } = request.params .arguments as any; try { const prData: Record = { head, base, title }; if (body) prData.body = body; const response = await giteaClient.post( `/repos/${owner}/${repo}/pulls`, prData, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error creating Pull Request: ${error.message}`, }, ], }; } } if (request.params.name === "gitea_search_repos") { const { query, limit = 10 } = request.params.arguments as any; try { const response = await giteaClient.get(`/repos/search`, { params: { q: query, limit }, }); return { content: [ { type: "text", text: JSON.stringify(response.data.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [{ type: "text", text: `Error: ${error.message}` }], }; } } // --- Wiki Handlers --- if (request.params.name === "gitea_list_wiki_pages") { const { owner, repo } = request.params.arguments as any; try { const response = await giteaClient.get( `/repos/${owner}/${repo}/wiki/pages`, ); const pages = (response.data || []).map((p: any) => ({ title: p.title, last_commit: p.last_commit?.message, })); return { content: [{ type: "text", text: JSON.stringify(pages, null, 2) }], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error listing wiki pages: ${error.message}` }, ], }; } } if (request.params.name === "gitea_get_wiki_page") { const { owner, repo, page_name } = request.params.arguments as any; try { const response = await giteaClient.get( `/repos/${owner}/${repo}/wiki/page/${encodeURIComponent(page_name)}`, ); const content = Buffer.from( response.data.content_base64 || "", "base64", ).toString("utf-8"); return { content: [ { type: "text", text: `# ${response.data.title}\n\n${content}` }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error fetching wiki page: ${error.message}` }, ], }; } } if (request.params.name === "gitea_create_wiki_page") { const { owner, repo, title, content, message } = request.params .arguments as any; try { const response = await giteaClient.post( `/repos/${owner}/${repo}/wiki/pages`, { title, content_base64: Buffer.from(content).toString("base64"), message: message || `Create wiki page: ${title}`, }, ); return { content: [ { type: "text", text: `Wiki page '${response.data.title}' created.` }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error creating wiki page: ${error.message}` }, ], }; } } if (request.params.name === "gitea_edit_wiki_page") { const { owner, repo, page_name, title, content, message } = request.params .arguments as any; try { const updateData: Record = { content_base64: Buffer.from(content).toString("base64"), message: message || `Update wiki page: ${page_name}`, }; if (title) updateData.title = title; const response = await giteaClient.patch( `/repos/${owner}/${repo}/wiki/pages/${encodeURIComponent(page_name)}`, updateData, ); return { content: [ { type: "text", text: `Wiki page '${response.data.title}' updated.` }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error updating wiki page: ${error.message}` }, ], }; } } // --- Package Handlers --- if (request.params.name === "gitea_list_packages") { const { owner, type, limit = 10 } = request.params.arguments as any; try { const params: Record = { limit }; if (type) params.type = type; const response = await giteaClient.get(`/packages/${owner}`, { params }); const packages = (response.data || []).map((p: any) => ({ name: p.name, type: p.type, version: p.version, created: p.created_at, })); return { content: [{ type: "text", text: JSON.stringify(packages, null, 2) }], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error listing packages: ${error.message}` }, ], }; } } if (request.params.name === "gitea_list_package_versions") { const { owner, type, name } = request.params.arguments as any; try { const response = await giteaClient.get( `/packages/${owner}/${type}/${encodeURIComponent(name)}`, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error listing package versions: ${error.message}`, }, ], }; } } // --- Release Handlers --- if (request.params.name === "gitea_list_releases") { const { owner, repo, limit = 10 } = request.params.arguments as any; try { const response = await giteaClient.get( `/repos/${owner}/${repo}/releases`, { params: { limit } }, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error listing releases: ${error.message}` }, ], }; } } if (request.params.name === "gitea_get_latest_release") { const { owner, repo } = request.params.arguments as any; try { const response = await giteaClient.get( `/repos/${owner}/${repo}/releases/latest`, ); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error fetching latest release: ${error.message}`, }, ], }; } } if (request.params.name === "gitea_create_release") { const { owner, repo, tag_name, name, body, draft = false, prerelease = false, } = request.params.arguments as any; try { const response = await giteaClient.post( `/repos/${owner}/${repo}/releases`, { tag_name, name, body, draft, prerelease, }, ); return { content: [ { type: "text", text: `Release '${response.data.name}' created: ${response.data.html_url}`, }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error creating release: ${error.message}` }, ], }; } } // --- Project Handlers --- if (request.params.name === "gitea_list_projects") { const { owner, repo } = request.params.arguments as any; try { // Gitea API: repo-level projects or user projects const url = repo ? `/repos/${owner}/${repo}/projects` : `/users/${owner}/projects`; const response = await giteaClient.get(url); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error listing projects: ${error.message}` }, ], }; } } if (request.params.name === "gitea_get_project_columns") { const { project_id } = request.params.arguments as any; try { const response = await giteaClient.get(`/projects/${project_id}/columns`); return { content: [ { type: "text", text: JSON.stringify(response.data, null, 2) }, ], }; } catch (error: any) { return { isError: true, content: [ { type: "text", text: `Error fetching project columns: ${error.message}`, }, ], }; } } throw new Error(`Unknown tool: ${request.params.name}`); }); // --- Resources & Subscriptions --- // We will expose a dynamic resource URI pattern: gitea://runs/{owner}/{repo}/{run_id} server.setRequestHandler(ListResourcesRequestSchema, async () => { return { resources: [ { uri: "gitea://runs", name: "Gitea Pipeline Runs", description: "Dynamic resource for subscribing to pipeline runs. Format: gitea://runs/{owner}/{repo}/{run_id}", mimeType: "application/json", }, ], }; }); server.setRequestHandler(ReadResourceRequestSchema, async (request) => { const uri = request.params.uri; const match = uri.match(/^gitea:\/\/runs\/([^/]+)\/([^/]+)\/(\d+)$/); if (!match) { throw new Error( `Invalid resource URI. Must be gitea://runs/{owner}/{repo}/{run_id}`, ); } const [, owner, repo, run_id] = match; try { const runResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run_id}`, ); const jobsResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run_id}/jobs`, ); const resourceContent = { run: runResponse.data, jobs: jobsResponse.data, }; // Update internal cache when read runStatusCache.set(uri, runResponse.data.status); return { contents: [ { uri, mimeType: "application/json", text: JSON.stringify(resourceContent, null, 2), }, ], }; } catch (error: any) { throw new Error(`Failed to read Gitea resource: ${error.message}`); } }); server.setRequestHandler(SubscribeRequestSchema, async (request) => { const uri = request.params.uri; if (!uri.startsWith("gitea://runs/")) { throw new Error("Only gitea://runs resources can be subscribed to"); } subscriptions.add(uri); console.error(`Client subscribed to ${uri}`); return {}; }); server.setRequestHandler(UnsubscribeRequestSchema, async (request) => { const uri = request.params.uri; subscriptions.delete(uri); console.error(`Client unsubscribed from ${uri}`); return {}; }); // The server polling mechanism that pushes updates to subscribed clients async function pollSubscriptions() { for (const uri of subscriptions) { const match = uri.match(/^gitea:\/\/runs\/([^/]+)\/([^/]+)\/(\d+)$/); if (!match) continue; const [, owner, repo, run_id] = match; try { const runResponse = await giteaClient.get( `/repos/${owner}/${repo}/actions/runs/${run_id}`, ); const currentStatus = runResponse.data.status; const prevStatus = runStatusCache.get(uri); // If status changed (e.g. running -> completed), notify client if (prevStatus !== currentStatus) { runStatusCache.set(uri, currentStatus); server.notification({ method: "notifications/resources/updated", params: { uri }, }); console.error( `Pushed update for ${uri}: ${prevStatus} -> ${currentStatus}`, ); // Auto-unsubscribe if completed/failed so we don't poll forever? // Let the client decide, or we can handle it here if requested. } } catch (e) { console.error(`Error polling subscription ${uri}:`, e); } } // Poll every 5 seconds setTimeout(pollSubscriptions, 5000); } async function run() { const isStdio = process.argv.includes("--stdio"); if (isStdio) { const { StdioServerTransport } = await import("@modelcontextprotocol/sdk/server/stdio.js"); const transport = new StdioServerTransport(); await server.connect(transport); console.error("Gitea MCP server is running on stdio"); } else { const app = express(); const transports = new Map(); // Middleware to log all requests for debugging app.use((req, _res, next) => { console.error(`${req.method} ${req.url}`); next(); }); app.get("/sse", async (req, res) => { const sessionId = crypto.randomUUID(); console.error(`New SSE connection: ${sessionId}`); const transport = new SSEServerTransport(`/message/${sessionId}`, res); transports.set(sessionId, transport); req.on("close", () => { console.error(`SSE connection closed: ${sessionId}`); transports.delete(sessionId); }); await server.connect(transport); }); app.post("/message/:sessionId", async (req, res) => { const { sessionId } = req.params; const transport = transports.get(sessionId); if (!transport) { console.error(`No transport found for session: ${sessionId}`); res.status(400).send("No active SSE connection for this session"); return; } await transport.handlePostMessage(req, res); }); const PORT = process.env.GITEA_MCP_PORT || 3001; app.listen(PORT, () => { console.error(`Gitea MCP server running on http://localhost:${PORT}/sse`); }); // Start the background poller only in SSE mode or if specifically desired pollSubscriptions(); } } run().catch((error) => { console.error("Fatal error:", error); process.exit(1); });