feat: add community contribution for MCP lab

6_mcp/community_contributions/PrasadGaitonde/6_MCP_GitHubProject/.claude/settings.local.json

@@ -0,0 +1,8 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(pip install *)",
+      "Bash(uv pip *)"
+    ]
+  }
+}

6_mcp/community_contributions/PrasadGaitonde/6_MCP_GitHubProject/.gitignore

@@ -0,0 +1,46 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+
+# Runtime data / Dependencies
+node_modules/
+jspm_packages/
+pnpm-store/
+.pnpm-debug.log*
+
+# Environments and Secrets (NEVER commit these!)
+.env
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+
+# Build outputs
+dist/
+build/
+out/
+
+# Python compiled files
+__pycache__/
+*.py[cod]
+*$py.class
+.venv/
+venv/
+ENV/
+
+# IDEs and Editors
+.idea/
+.vscode/
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.swp
+
+# System Files
+.DS_Store
+Thumbs.db
+uv.lock
+.uvlock
+*.uvlock

6_mcp/community_contributions/PrasadGaitonde/6_MCP_GitHubProject/database.py

@@ -0,0 +1,34 @@
+import sqlite3
+import os
+from datetime import datetime
+
+DB_PATH = os.path.join(os.path.dirname(__file__), "github_agent.db")
+
+def init_db():
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS logs (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            timestamp TEXT,
+            agent_name TEXT,
+            log_type TEXT,
+            message TEXT
+        )
+    ''')
+    conn.commit()
+    conn.close()
+
+def write_log(agent_name: str, log_type: str, message: str):
+    timestamp = datetime.now().isoformat()
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute('''
+        INSERT INTO logs (timestamp, agent_name, log_type, message)
+        VALUES (?, ?, ?, ?)
+    ''', (timestamp, agent_name, log_type, message))
+    conn.commit()
+    conn.close()
+
+# Initialize the database on import
+init_db()

6_mcp/community_contributions/PrasadGaitonde/6_MCP_GitHubProject/github_client.py

@@ -0,0 +1,120 @@
+import os
+import json
+import asyncio
+import contextlib
+from typing import List, Optional, Any
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from tracers import LogTracer
+from database import write_log
+from dotenv import load_dotenv
+
+# Path to the .env file in the parent's parent directory
+ENV_PATH = "/home/prasad/projects/agents/.env"
+load_dotenv(ENV_PATH)
+
+# Workaround for the token name in the .env file
+if os.getenv("GITLAB_PERSONAL_ACCESS_TOKEN") and not os.getenv("GITHUB_TOKEN"):
+    os.environ["GITHUB_TOKEN"] = os.getenv("GITLAB_PERSONAL_ACCESS_TOKEN")
+
+# The path to the github_server.py
+SERVER_PATH = os.path.join(os.path.dirname(__file__), "github_server.py")
+
+class GitHubClientBridge:
+    def __init__(self):
+        self.session: Optional[ClientSession] = None
+        self.server_params = StdioServerParameters(
+            command="uv",
+            args=["run", SERVER_PATH],
+            env={**os.environ}
+        )
+        self.tracer = LogTracer()
+        self._exit_stack = contextlib.AsyncExitStack()
+
+    async def connect(self):
+        """Initialize the stdio client session."""
+        try:
+            stdio_transport = await self._exit_stack.enter_async_context(stdio_client(self.server_params))
+            self.session = await self._exit_stack.enter_async_context(ClientSession(stdio_transport[0], stdio_transport[1]))
+            await self.session.initialize()
+            self.tracer.info("Successfully connected to GitHub MCP server")
+        except Exception as e:
+            self.tracer.error(f"Failed to connect to GitHub MCP server: {e}")
+            raise
+
+    async def disconnect(self):
+        """Shutdown the session."""
+        await self._exit_stack.aclose()
+        self.session = None
+
+    async def get_github_tools_openai(self) -> List[dict]:
+        """
+        Retrieve MCP tools and convert them to OpenAI-compatible function schemas.
+        """
+        if not self.session:
+            return []
+
+        mcp_tools = await self.session.list_tools()
+        openai_tools = []
+
+        for tool in mcp_tools.tools:
+            openai_tools.append({
+                "type": "function",
+                "function": {
+                    "name": tool.name,
+                    "description": tool.description,
+                    "parameters": tool.inputSchema
+                }
+            })
+        return openai_tools
+
+    async def read_github_resource(self, uri: str) -> str:
+        """
+        Retrieve data for a given GitHub MCP resource URI.
+        """
+        if not self.session:
+            raise RuntimeError("Client session not connected")
+
+        # Wrap in tracing
+        try:
+            resource = await self.session.read_resource(uri)
+            content = resource.content[0].text if resource.content else ""
+            self.tracer.resource(uri, content)
+            return content
+        except Exception as e:
+            self.tracer.error(f"Error reading resource {uri}: {e}")
+            raise
+
+    async def execute_tool(self, tool_name: str, arguments: dict) -> Any:
+        """
+        Execute a tool on the GitHub MCP server.
+        """
+        if not self.session:
+            raise RuntimeError("Client session not connected")
+
+        # Wrap in tracing
+        try:
+            result = await self.session.call_tool(tool_name, arguments)
+            text_result = result.content[0].text if result.content else ""
+            self.tracer.tool(tool_name, arguments, text_result)
+            return text_result
+        except Exception as e:
+            self.tracer.error(f"Error executing tool {tool_name}: {e}")
+            raise
+
+# Singleton for easy access
+github_bridge = GitHubClientBridge()
+
+if __name__ == "__main__":
+    async def main():
+        bridge = GitHubClientBridge()
+        try:
+            await bridge.connect()
+            tools = await bridge.get_github_tools_openai()
+            print(f"Retrieved {len(tools)} tools")
+            for tool in tools:
+                print(f"- {tool['function']['name']}: {tool['function']['description']}")
+        finally:
+            await bridge.disconnect()
+
+    asyncio.run(main())

6_mcp/community_contributions/PrasadGaitonde/6_MCP_GitHubProject/github_server.py

@@ -0,0 +1,157 @@
+import os
+from typing import List, Optional
+from fastmcp import FastMCP
+from github import Github, GithubException
+from dotenv import load_dotenv
+
+# Path to the .env file in the parent's parent directory
+ENV_PATH = "/home/prasad/projects/agents/.env"
+load_dotenv(ENV_PATH)
+
+# Workaround for the token name in the .env file
+if os.getenv("GITLAB_PERSONAL_ACCESS_TOKEN") and not os.getenv("GITHUB_TOKEN"):
+    os.environ["GITHUB_TOKEN"] = os.getenv("GITLAB_PERSONAL_ACCESS_TOKEN")
+
+# Initialize FastMCP server
+mcp = FastMCP("GitHub Agent")
+
+# Initialize GitHub client
+# We assume GITHUB_TOKEN is set in the environment
+def get_github_client():
+    token = os.getenv("GITHUB_TOKEN")
+    if not token:
+        raise RuntimeError("GITHUB_TOKEN environment variable is not set")
+    return Github(token)
+
+@mcp.tool()
+def get_repo_structure(owner: str, repo: str, path: str = "") -> str:
+    """
+    Provides a tree view of the repository at the specified path.
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        contents = repository.get_contents(path)
+
+        structure = []
+        for item in contents:
+            prefix = "📁" if item.type == "dir" else "📄"
+            structure.append(f"{prefix} {item.path}")
+
+        return "\n".join(structure) if structure else "Directory is empty."
+    except GithubException as e:
+        return f"Error fetching repo structure: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+@mcp.tool()
+def read_file(owner: str, repo: str, path: str) -> str:
+    """
+    Reads the content of a specific file from a GitHub repository.
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        content = repository.get_contents(path).decoded_content
+        return content.decode("utf-8")
+    except GithubException as e:
+        return f"Error reading file: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+@mcp.tool()
+def create_issue(owner: str, repo: str, title: str, body: str) -> str:
+    """
+    Creates a new issue in the specified GitHub repository.
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        issue = repository.create_issue(title=title, body=body)
+        return f"Issue created successfully: {issue.html_url}"
+    except GithubException as e:
+        return f"Error creating issue: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+@mcp.tool()
+def list_prs(owner: str, repo: str, state: str = "open") -> str:
+    """
+    Lists pull requests in a GitHub repository based on the state (open, closed, all).
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        pulls = repository.get_pulls(state=state, sort="created", direction="descending")
+
+        pr_list = []
+        for pr in pulls:
+            pr_list.append(f"#{pr.number}: {pr.title} (Author: {pr.user.login}) - {pr.html_url}")
+
+        return "\n".join(pr_list) if pr_list else "No pull requests found."
+    except GithubException as e:
+        return f"Error listing PRs: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+@mcp.tool()
+def create_pull_request(owner: str, repo: str, title: str, head: str, base: str, body: str) -> str:
+    """
+    Creates a new pull request in the specified GitHub repository.
+    - head: The name of the branch where your changes are implemented.
+    - base: The name of the branch you want the changes merged into.
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        pr = repository.create_pull(title=title, body=body, head=head, base=base)
+        return f"Pull request created successfully: {pr.html_url}"
+    except GithubException as e:
+        return f"Error creating pull request: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+@mcp.tool()
+def search_code(query: str) -> str:
+    """
+    Searches for code across GitHub repositories based on the query.
+    """
+    try:
+        g = get_github_client()
+        results = g.search_code(query=query)
+
+        search_results = []
+        for item in results:
+            search_results.append(f"File: {item.path} in {item.repository.full_name} - {item.html_url}")
+
+        return "\n".join(search_results) if search_results else "No code matches found."
+    except GithubException as e:
+        return f"Error searching code: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+# Resources
+@mcp.resource("github://repo/{owner}/{repo}/files")
+def get_repo_files(owner: str, repo: str) -> str:
+    """
+    Returns the root directory listing of a repository.
+    """
+    return get_repo_structure(owner, repo)
+
+@mcp.resource("github://issue/{owner}/{repo}/{issue_number}")
+def get_issue_details(owner: str, repo: str, issue_number: int) -> str:
+    """
+    Returns details of a specific GitHub issue.
+    """
+    try:
+        g = get_github_client()
+        repository = g.get_repo(f"{owner}/{repo}")
+        issue = repository.get_issue(number=issue_number)
+        return f"Title: {issue.title}\nState: {issue.state}\nBody: {issue.body}"
+    except GithubException as e:
+        return f"Error fetching issue details: {e}"
+    except Exception as e:
+        return f"An unexpected error occurred: {e}"
+
+if __name__ == "__main__":
+    mcp.run()