github_mcp_agent/github_agent.py at main · AzaRKazar/github_mcp_agent · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import asyncio
import os
import streamlit as st
from textwrap import dedent
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.ollama import Ollama
from agno.tools.mcp import MCPTools
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import nest_asyncio
nest_asyncio.apply()


# Load env vars
load_dotenv()

# Page config
st.set_page_config(page_title="🐙 GitHub MCP Agent", page_icon="🐙", layout="wide")

# Title + description
st.markdown("## 🐙 GitHub MCP Agent")
st.write("Explore GitHub repositories with **natural language** using the **Model Context Protocol (MCP)**.")

# Sidebar for authentication
with st.sidebar:
    st.header("🔑 Authentication")
    github_token = st.text_input(
        "GitHub Token", type="password",
        value=os.getenv("GITHUB_TOKEN", ""),
        help="Create a token with repo scope at github.com/settings/tokens"
    )
    if github_token:
        os.environ["GITHUB_TOKEN"] = github_token

    st.markdown("---")
    st.markdown("### Example Queries")
    st.markdown("**Issues**\n- Show me issues by label\n- What issues are being actively discussed?")
    st.markdown("**Pull Requests**\n- What PRs need review?\n- Show me recent merged PRs")
    st.markdown("**Repository**\n- Show repository health metrics\n- Show repository activity patterns")
    st.caption("Tip: Always specify the repository if not included in the query.")

# Repo + query input
col1, col2 = st.columns([3, 1])
with col1:
    repo = st.text_input("Repository", value="AzaRKazar/Athlete-Injury-Risk-Analyzer", help="Format: owner/repo")
with col2:
    query_type = st.selectbox("Query Type", ["Issues", "Pull Requests", "Repository Activity", "Custom"])

# Predefined query templates
templates = {
    "Issues": f"List open issues in {repo} with labels and comment counts",
    "Pull Requests": f"Show me recent merged PRs in {repo}",
    "Repository Activity": f"Summarize code quality and commit activity trends in {repo}",
    "Custom": ""
}
query = st.text_area("Your Query", value=templates.get(query_type, ""))

# Async function
async def run_github_agent(message):
    if not os.getenv("GITHUB_TOKEN"):
        return "⚠️ Error: GitHub token not provided."

    try:
        server_params = StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
        )
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                mcp_tools = MCPTools(session=session)
                await mcp_tools.initialize()

                # Agent w/ Ollama reasoning
                agent = Agent(
                    model=Ollama(id="llama3.2:latest"),  # Local reasoning model
                    tools=[mcp_tools],
                    instructions=dedent("""\
                        You are a GitHub analysis assistant.
                        - Use MCP GitHub data to answer queries.
                        - Summarize results clearly with markdown.
                        - Use tables for issues/PRs.
                        - Add links to GitHub resources where relevant.
                    """),
                    markdown=True,
                )

                response = await agent.arun(message)
                return response.content or "⚠️ No results returned."
    except Exception as e:
        return f"🚨 MCP Error: {str(e)}"

# Run button
if st.button("🚀 Run Query", type="primary", use_container_width=True):
    if not github_token:
        st.error("Please enter your GitHub token in the sidebar")
    elif not query:
        st.error("Please enter a query")
    else:
        with st.spinner("🔎 Analyzing GitHub repository..."):
            full_query = query if repo in query else f"{query} in {repo}"
            result = asyncio.run(run_github_agent(full_query))

        st.markdown("### 📊 Results")
        st.markdown(result)

# Help text
st.info(
    """
    **How to use this app**
    1. Enter your GitHub token in the sidebar
    2. Enter a repository (e.g., `AzaRKazar/Athlete-Injury-Risk-Analyzer`)
    3. Choose a query type or write your own
    4. Click **Run Query** to analyze GitHub data

    ⚠️ Requires Node.js installed for `npx` to work.
    """
)

# Footer
st.markdown("---")
st.caption("Built with ❤️ : Azar")