viraja1's picture
Create app.py
dff9abc verified
Raw
History Blame Contribute Delete
30.9 kB
"""MCP server exposing GitHub tooling for ChatGPT apps."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from mcp.server.fastmcp import FastMCP
from github_client import GitHubAPIError, GitHubClient
server = FastMCP("github-gpt-tools")
_client = GitHubClient()
HTML_MIME_TYPE = "text/html+skybridge"
RESOURCE_URIS = {
"repo_info": "ui://github/repo_info.html",
"issues": "ui://github/issues.html",
"pulls": "ui://github/pulls.html",
"readme": "ui://github/readme.html",
"contributors": "ui://github/contributors.html",
"files": "ui://github/files.html",
"search": "ui://github/search.html",
}
def _tool_meta(resource_uri: str) -> Dict[str, Any]:
return {
"openai/outputTemplate": resource_uri,
"openai/resultCanProduceWidget": True,
"openai/widgetAccessible": True,
}
@server.resource(
RESOURCE_URIS["repo_info"],
title="Repository Overview",
description="Summarize repository metadata with links and stats.",
mime_type=HTML_MIME_TYPE,
)
def repo_info_resource() -> str:
return """
<div id="repo-info"></div>
<script>
let lastRepo;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate) {
lastRepo = candidate;
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentRepo() {
if (lastRepo === undefined) {
if (!captureInitial()) {
lastRepo = {};
}
}
return lastRepo || {};
}
function formatNumber(value) {
if (typeof value !== "number") return value ?? "0";
return value.toLocaleString();
}
function render() {
const repo = currentRepo();
const root = document.getElementById("repo-info");
if (!root) return;
if (!repo.full_name) {
root.innerHTML = "<p>No repository selected yet.</p>";
return;
}
root.innerHTML = `
<h2 style="margin-bottom:0.4rem;">${repo.full_name}</h2>
<p style="margin:0 0 0.5rem;">${repo.description || "No description provided."}</p>
<div style="display:flex;gap:1rem;font-size:14px;margin:0.5rem 0 1rem;">
<span>⭐ ${formatNumber(repo.stargazers_count || 0)}</span>
<span>🍴 ${formatNumber(repo.forks_count || 0)}</span>
<span>❗ ${formatNumber(repo.open_issues_count || 0)}</span>
</div>
<div style="display:flex;flex-wrap:wrap;gap:0.4rem;margin-bottom:0.75rem;">
${(repo.topics || []).map(topic => `<span style="padding:2px 8px;border-radius:999px;background:#f1f5f9;font-size:12px;">${topic}</span>`).join("")}
</div>
<a href="${repo.html_url}" target="_blank" style="font-weight:500;">Open on GitHub ↗</a>
`;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["issues"],
title="Issue List",
description="Show currently fetched GitHub issues.",
mime_type=HTML_MIME_TYPE,
)
def issues_resource() -> str:
return """
<div id="issues-list"></div>
<script>
let lastIssues;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node !== undefined) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate !== undefined) {
if (Array.isArray(candidate)) {
lastIssues = candidate;
} else if (candidate) {
lastIssues = [candidate];
} else {
lastIssues = [];
}
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentIssues() {
if (lastIssues === undefined) {
if (!captureInitial()) {
lastIssues = [];
}
}
if (Array.isArray(lastIssues)) return lastIssues;
if (lastIssues && typeof lastIssues === "object") return [lastIssues];
return [];
}
function render() {
const items = currentIssues();
const root = document.getElementById("issues-list");
if (!root) return;
if (!items.length) {
root.innerHTML = "<p>No issues returned.</p>";
return;
}
let html = "<h3>Issues</h3><ul style='margin:0;padding-left:1.2rem;'>";
for (const issue of items) {
html += `
<li style="margin-bottom:0.4rem;">
<a href="${issue.html_url}" target="_blank">
#${issue.number} — ${issue.title}
</a>
<div style="font-size:12px;color:#475569;">${issue.state} · ${issue.author || "unknown"}</div>
</li>
`;
}
html += "</ul>";
root.innerHTML = html;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["pulls"],
title="Pull Request List",
description="Summaries for GitHub pull requests.",
mime_type=HTML_MIME_TYPE,
)
def pull_requests_resource() -> str:
return """
<div id="pulls-list"></div>
<script>
let lastPulls;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node !== undefined) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate !== undefined) {
if (Array.isArray(candidate)) {
lastPulls = candidate;
} else if (candidate) {
lastPulls = [candidate];
} else {
lastPulls = [];
}
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentPulls() {
if (lastPulls === undefined) {
if (!captureInitial()) {
lastPulls = [];
}
}
if (Array.isArray(lastPulls)) return lastPulls;
if (lastPulls && typeof lastPulls === "object") return [lastPulls];
return [];
}
function render() {
const pulls = currentPulls();
const root = document.getElementById("pulls-list");
if (!root) return;
if (!pulls.length) {
root.innerHTML = "<p>No pull requests returned.</p>";
return;
}
let html = "<h3>Pull Requests</h3><ul style='margin:0;padding-left:1.2rem;'>";
for (const pr of pulls) {
html += `
<li style="margin-bottom:0.5rem;">
<a href="${pr.html_url}" target="_blank">#${pr.number} — ${pr.title}</a>
<div style="font-size:12px;color:#475569;">${pr.state}${pr.draft ? " · draft" : ""} · ${pr.author || "unknown"}</div>
<div style="font-size:12px;color:#475569;">${pr.head || ""} ⇢ ${pr.base || ""}</div>
</li>
`;
}
html += "</ul>";
root.innerHTML = html;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["readme"],
title="README Viewer",
description="Render README content as preformatted text.",
mime_type=HTML_MIME_TYPE,
)
def readme_resource() -> str:
return """
<div id="readme-viewer"></div>
<script>
let lastReadme;
function pickFrom(detail) {
if (!detail || typeof detail !== "object") return undefined;
if (detail.toolResult) return detail.toolResult;
if (detail.toolOutput?.result) return detail.toolOutput.result;
if (detail.payload?.data) return detail.payload.data;
if (detail.result) return detail.result;
if (detail.toolOutput) return detail.toolOutput;
if (detail.payload) return detail.payload;
return undefined;
}
function updateFromEvent(event) {
if (!event) return false;
const detail = event.detail || {};
const candidate = pickFrom(detail) ?? pickFrom(detail.globals);
if (candidate !== undefined) {
lastReadme = candidate;
return true;
}
return false;
}
function currentReadme() {
if (lastReadme === undefined) {
lastReadme = {};
}
return lastReadme || {};
}
function render() {
const result = currentReadme();
const root = document.getElementById("readme-viewer");
if (!root) return;
const hasError = Boolean(result.error);
const text = result.content || (hasError ? "" : "README not available.");
const errorHtml = hasError
? `<div style="margin:0.5rem 0;padding:0.5rem;background:#fee2e2;border:1px solid #fecaca;border-radius:6px;color:#b91c1c;font-size:13px;">${result.error}</div>`
: "";
root.innerHTML = `
<h3>README (${result.path || "latest"})</h3>
${errorHtml}
<pre style="background:#0f172a0d;border:1px solid #cbd5f5;border-radius:6px;padding:0.75rem;white-space:pre-wrap;overflow:auto;max-height:360px;"></pre>
`;
const pre = root.querySelector("pre");
if (pre) {
pre.textContent = text;
}
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["contributors"],
title="Contributor List",
description="Display contributors and their commit counts.",
mime_type=HTML_MIME_TYPE,
)
def contributors_resource() -> str:
return """
<div id="contributors-list"></div>
<script>
let lastContributors;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node !== undefined) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate !== undefined) {
if (Array.isArray(candidate)) {
lastContributors = candidate;
} else if (candidate) {
lastContributors = [candidate];
} else {
lastContributors = [];
}
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentContributors() {
if (lastContributors === undefined) {
if (!captureInitial()) {
lastContributors = [];
}
}
if (Array.isArray(lastContributors)) return lastContributors;
if (lastContributors && typeof lastContributors === "object") return [lastContributors];
return [];
}
function render() {
const people = Array.isArray(currentContributors()) ? currentContributors() : [];
const root = document.getElementById("contributors-list");
if (!root) return;
if (!people.length) {
root.innerHTML = "<p>No contributors returned.</p>";
return;
}
let html = "<h3>Top Contributors</h3><ul style='list-style:none;margin:0;padding:0;display:flex;flex-direction:column;gap:0.35rem;'>";
for (const person of people) {
html += `
<li style="display:flex;align-items:center;gap:0.5rem;">
<span style="font-weight:600;">${person.login || "unknown"}</span>
<span style="font-size:12px;color:#475569;">${person.contributions || 0} commits</span>
<a href="${person.html_url}" target="_blank" style="font-size:12px;">profile ↗</a>
</li>
`;
}
html += "</ul>";
root.innerHTML = html;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["files"],
title="Repository Files",
description="Visualize the returned Git tree entries.",
mime_type=HTML_MIME_TYPE,
)
def files_resource() -> str:
return """
<div id="files-list"></div>
<script>
let lastFiles;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node !== undefined) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate !== undefined) {
if (Array.isArray(candidate)) {
lastFiles = candidate;
} else if (candidate) {
lastFiles = [candidate];
} else {
lastFiles = [];
}
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentFiles() {
if (lastFiles === undefined) {
if (!captureInitial()) {
lastFiles = [];
}
}
if (Array.isArray(lastFiles)) return lastFiles;
if (lastFiles && typeof lastFiles === "object") return [lastFiles];
return [];
}
function render() {
const nodes = Array.isArray(currentFiles()) ? currentFiles() : [];
const root = document.getElementById("files-list");
if (!root) return;
if (!nodes.length) {
root.innerHTML = "<p>No files returned.</p>";
return;
}
let html = "<h3>Files</h3><ul style='margin:0;padding-left:1.2rem;'>";
for (const node of nodes) {
const icon = node.type === "tree" ? "📁" : "📄";
const sizeInfo = node.size ? ` <span style='color:#475569;font-size:12px;'>${node.size} bytes</span>` : "";
const link = node.html_url ? `<a href='${node.html_url}' target='_blank'>${node.path}</a>` : node.path;
html += `<li style="margin-bottom:0.35rem;">${icon} ${link}${sizeInfo}</li>`;
}
html += "</ul>";
root.innerHTML = html;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
@server.resource(
RESOURCE_URIS["search"],
title="Repository Search Results",
description="Render search results with metadata.",
mime_type=HTML_MIME_TYPE,
)
def search_resource() -> str:
return """
<div id="search-results"></div>
<script>
let lastSearch;
const PATHS = [
["toolResult"],
["toolOutput","result"],
["toolOutput"],
["result"],
["payload","data"],
["payload"],
["globals","toolOutput","result"],
["globals","toolOutput"],
["globals","widget","props","result"],
];
function pick(data) {
if (!data || typeof data !== "object") return undefined;
for (const path of PATHS) {
let node = data;
for (const key of path) {
if (!node || typeof node !== "object") {
node = undefined;
break;
}
node = node[key];
}
if (node !== undefined) return node;
}
return undefined;
}
function handle(detail) {
const candidate = pick(detail) ?? pick(detail?.globals);
if (candidate !== undefined) {
if (Array.isArray(candidate)) {
lastSearch = candidate;
} else if (candidate) {
lastSearch = [candidate];
} else {
lastSearch = [];
}
return true;
}
return false;
}
function updateFromEvent(event) {
if (!event) return false;
return handle(event.detail || {});
}
function captureInitial() {
return handle(window.openai || {});
}
function currentSearch() {
if (lastSearch === undefined) {
if (!captureInitial()) {
lastSearch = [];
}
}
if (Array.isArray(lastSearch)) return lastSearch;
if (lastSearch && typeof lastSearch === "object") return [lastSearch];
return [];
}
function render() {
const repos = Array.isArray(currentSearch()) ? currentSearch() : [];
const root = document.getElementById("search-results");
if (!root) return;
if (!repos.length) {
root.innerHTML = "<p>No repositories returned for this query.</p>";
return;
}
let html = "<h3>Search Results</h3><ul style='margin:0;padding-left:1.2rem;'>";
for (const repo of repos) {
html += `
<li style="margin-bottom:0.6rem;">
<a href="${repo.html_url}" target="_blank">${repo.full_name}</a>
<div style="font-size:12px;color:#475569;">⭐ ${repo.stargazers_count || 0} · Updated ${repo.updated_at || "recently"}</div>
<div style="font-size:13px;color:#1f2933;margin-top:0.2rem;">${repo.description || ""}</div>
</li>
`;
}
html += "</ul>";
root.innerHTML = html;
}
render();
const eventNames = ["openai:tool_result","openai:set_globals","openai:tool_output","openai:result","openai:response"];
for (const name of eventNames) {
window.addEventListener(name, (event) => {
const changed = updateFromEvent(event);
if (changed) render();
});
}
</script>
"""
def _summarize_repo(data: Dict[str, Any]) -> Dict[str, Any]:
return {
"id": data.get("id"),
"full_name": data.get("full_name"),
"description": data.get("description"),
"visibility": data.get("visibility"),
"default_branch": data.get("default_branch"),
"language": data.get("language"),
"license": (data.get("license") or {}).get("spdx_id") if data.get("license") else None,
"stargazers_count": data.get("stargazers_count"),
"forks_count": data.get("forks_count"),
"open_issues_count": data.get("open_issues_count"),
"html_url": data.get("html_url"),
"topics": data.get("topics", []),
"updated_at": data.get("updated_at"),
}
def _summarize_issue(issue: Dict[str, Any]) -> Dict[str, Any]:
return {
"number": issue.get("number"),
"title": issue.get("title"),
"state": issue.get("state"),
"html_url": issue.get("html_url"),
"author": (issue.get("user") or {}).get("login"),
"comments": issue.get("comments"),
"labels": [label.get("name") for label in issue.get("labels", [])],
"created_at": issue.get("created_at"),
"updated_at": issue.get("updated_at"),
"assignees": [assignee.get("login") for assignee in issue.get("assignees", [])],
}
def _summarize_pr(pr: Dict[str, Any]) -> Dict[str, Any]:
return {
"number": pr.get("number"),
"title": pr.get("title"),
"state": pr.get("state"),
"draft": pr.get("draft"),
"html_url": pr.get("html_url"),
"author": (pr.get("user") or {}).get("login"),
"created_at": pr.get("created_at"),
"updated_at": pr.get("updated_at"),
"merged_at": pr.get("merged_at"),
"head": (pr.get("head") or {}).get("ref"),
"base": (pr.get("base") or {}).get("ref"),
}
def _summarize_contributor(contributor: Dict[str, Any]) -> Dict[str, Any]:
return {
"login": contributor.get("login"),
"contributions": contributor.get("contributions"),
"html_url": contributor.get("html_url"),
"type": contributor.get("type"),
}
def _handle_errors(coro):
async def wrapper(*args, **kwargs):
try:
return await coro(*args, **kwargs)
except GitHubAPIError as exc:
return {"error": str(exc)}
return wrapper
@server.tool(meta=_tool_meta(RESOURCE_URIS["repo_info"]))
async def get_repo_info(owner: str, repo: str) -> Dict[str, Any]:
"""Return metadata for a repository."""
data = await _client.get_repo_info(owner, repo)
return _summarize_repo(data)
@server.tool(meta=_tool_meta(RESOURCE_URIS["issues"]))
async def list_issues(
owner: str,
repo: str,
state: str = "open",
labels: Optional[List[str]] = None,
limit: int = 25,
include_pull_requests: bool = False,
) -> List[Dict[str, Any]]:
"""List issues in a repository."""
issues = await _client.list_issues(owner, repo, state, labels, limit, include_pull_requests)
return [_summarize_issue(issue) for issue in issues]
@server.tool(meta=_tool_meta(RESOURCE_URIS["pulls"]))
async def list_pull_requests(owner: str, repo: str, state: str = "open", limit: int = 25) -> List[Dict[str, Any]]:
"""List pull requests for a repository."""
pulls = await _client.list_pull_requests(owner, repo, state, limit)
return [_summarize_pr(pr) for pr in pulls]
@server.tool(meta=_tool_meta(RESOURCE_URIS["readme"]))
async def get_readme(owner: str, repo: str, ref: Optional[str] = None) -> Dict[str, Any]:
"""Fetch the README contents for a repository."""
return await _client.get_readme(owner, repo, ref)
@server.tool(meta=_tool_meta(RESOURCE_URIS["contributors"]))
async def list_contributors(owner: str, repo: str, limit: int = 25, include_anonymous: bool = False) -> List[Dict[str, Any]]:
"""Return contributors for a repository."""
contributors = await _client.list_contributors(owner, repo, limit, include_anonymous)
return [_summarize_contributor(contributor) for contributor in contributors]
@server.tool(meta=_tool_meta(RESOURCE_URIS["files"]))
async def list_files(
owner: str,
repo: str,
ref: str = "main",
directory: Optional[str] = None,
limit: int = 200,
) -> List[Dict[str, Any]]:
"""List files (tree entries) for a branch or tag."""
entries = await _client.list_files(owner, repo, ref, directory, limit)
base_url = f"https://github.com/{owner}/{repo}"
enriched: List[Dict[str, Any]] = []
for entry in entries:
path = entry.get("path")
if not path:
enriched.append(entry)
continue
entry_type = entry.get("type")
html_url = f"{base_url}/tree/{ref}/{path}" if entry_type == "tree" else f"{base_url}/blob/{ref}/{path}"
enriched.append({**entry, "html_url": html_url})
return enriched
@server.tool(meta=_tool_meta(RESOURCE_URIS["search"]))
async def search_repos(query: str, sort: Optional[str] = None, order: str = "desc", limit: int = 25) -> List[Dict[str, Any]]:
"""Search repositories using the GitHub search API."""
data = await _client.search_repositories(query, sort, order, limit)
return [_summarize_repo(repo) for repo in data]
if __name__ == "__main__":
server.run(transport="streamable-http")