MCP com FastMCP (2/4): camada de transporte, contexto e resources

mcp.run(
transport="stdio")

mcp.run(
transport="streamable-http",
host="0.0.0.0",port=8000,
)

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
 
# Create FastMCP server
mcp = FastMCP(
    name="GitHubMCP",
    instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
    include_tags={"public"},
    exclude_tags={"first_issue"}
)
 
sub_mcp = FastMCP(
    name="SubMCP",
)
 
@mcp.tool(tags={"public", "production"})
async def list_repository_issues(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 10 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=10"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"private", "development"})
async def list_more_repository_issues(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 100 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=100"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"public", "first_issue"})
async def first_repository_issue(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Gets the first issue ever created in a GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list containing information about the first issue created
    """
    # Get the first issue by sorting by creation date in ascending order
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;per_page=1"
    print(f"Fetching first issue from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No issues found for this repository.")
                return [{"message": "No issues found for this repository."}]
 
            first_issue = issues_data[0]
            
            # Create a detailed summary of the first issue
            summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
            if first_issue.get('comments', 0) &gt; 0:
                summary += f" ({first_issue.get('comments')} comments)"
            
            issue_info = {
                "number": first_issue.get("number"),
                "title": first_issue.get("title"),
                "user": first_issue.get("user", {}).get("login"),
                "url": first_issue.get("html_url"),
                "state": first_issue.get("state"),
                "comments": first_issue.get("comments"),
                "created_at": first_issue.get("created_at"),
                "updated_at": first_issue.get("updated_at"),
                "body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) &gt; 500 else first_issue.get("body", ""),
                "summary": summary
            }
            
            print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")
            
            # Add context information
            result = {
                "repository": f"{owner}/{repo_name}",
                "note": "This is the very first issue created in this repository",
                "first_issue": issue_info
            }
            
            return [result]
            
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; str:
    """
    Returns a simple greeting.
    """
    return "Hello, world!"
 
mcp.mount("sub_mcp", sub_mcp)
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server, run with uv run client.py http://localhost:8000/mcp
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8000,
    )
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github_server.py
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile client_MCP/client.py
 
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
 
# Load environment variables from .env file
load_dotenv()
 
class FastMCPClient:
    """
    FastMCP client that integrates with Claude to process user queries
    and use tools exposed by a FastMCP server.
    """
    
    def __init__(self):
        """Initialize the FastMCP client with Anthropic and resource management."""
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
        self.client = None
        
    async def connect_to_server(self, server_url: str):
        """
        Connect to the specified FastMCP server via HTTP.
        
        Args:
            server_url: URL of the HTTP server (e.g., "http://localhost:8000")
        """
        print(f"🔗 Connecting to FastMCP HTTP server: {server_url}")
        
        # Create FastMCP client for HTTP connection using SSE transport
        self.client = Client(server_url)
        # Note: FastMCP Client automatically detects HTTP URLs and uses SSE transport
        
        print("✅ Client created successfully")
        
    async def list_available_tools(self):
        """List available tools in the FastMCP server."""
        try:
            # Get list of tools from the server using FastMCP context
            async with self.client as client:
                tools = await client.list_tools()
                
                if tools:
                    print(f"
🛠️  Available tools ({len(tools)}):")
                    print("=" * 50)
                    
                    for tool in tools:
                        print(f"📋 {tool.name}")
                        if tool.description:
                            print(f"   Description: {tool.description}")
                        
                        # Show parameters if available
                        if hasattr(tool, 'inputSchema') and tool.inputSchema:
                            if 'properties' in tool.inputSchema:
                                params = list(tool.inputSchema['properties'].keys())
                                print(f"   Parameters: {', '.join(params)}")
                        print()
                else:
                    print("⚠️  No tools found in the server")
                    
        except Exception as e:
            print(f"❌ Error listing tools: {str(e)}")
 
    async def process_query(self, query: str) -&gt; str:
        """
        Process a user query, interacting with Claude and FastMCP tools.
        
        Args:
            query: User query
            
        Returns:
            str: Final processed response
        """
        try:
            # Use FastMCP context for all operations
            async with self.client as client:
                # Get available tools
                tools_list = await client.list_tools()
                
                # Prepare tools for Claude in correct format
                claude_tools = []
                for tool in tools_list:
                    claude_tool = {
                        "name": tool.name,
                        "description": tool.description or f"Tool {tool.name}",
                        "input_schema": tool.inputSchema or {"type": "object", "properties": {}}
                    }
                    claude_tools.append(claude_tool)
                
                # Create initial message for Claude
                messages = [
                    {
                        "role": "user",
                        "content": query
                    }
                ]
                
                # First call to Claude
                response = self.anthropic.messages.create(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=6000,
                    messages=messages,
                    tools=claude_tools if claude_tools else None
                )
                
                # Process Claude's response
                response_text = ""
                
                for content_block in response.content:
                    if content_block.type == "text":
                        response_text += content_block.text
                        
                    elif content_block.type == "tool_use":
                        # Claude wants to use a tool
                        tool_name = content_block.name
                        tool_args = content_block.input
                        tool_call_id = content_block.id
                        
                        print(f"🔧 Claude wants to use: {tool_name}")
                        print(f"📝 Arguments: {tool_args}")
                        
                        try:
                            # Execute tool on the FastMCP server
                            tool_result = await client.call_tool(tool_name, tool_args)
                            
                            print(f"✅ Tool executed successfully")
                            
                            # Add tool result to the conversation
                            messages.append({
                                "role": "assistant",
                                "content": response.content
                            })
                            
                            # Format result for Claude
                            if tool_result:
                                # Convert result to string format for Claude
                                result_content = str(tool_result)
                                
                                messages.append({
                                    "role": "user",
                                    "content": [{
                                        "type": "tool_result",
                                        "tool_use_id": tool_call_id,
                                        "content": f"Tool result: {result_content}"
                                    }]
                                })
                            else:
                                messages.append({
                                    "role": "user",
                                    "content": [{
                                        "type": "tool_result",
                                        "tool_use_id": tool_call_id,
                                        "content": "Tool executed without response content"
                                    }]
                                })
                            
                            # Second call to Claude with the tool result
                            final_response = self.anthropic.messages.create(
                                model="claude-3-5-sonnet-20241022",
                                max_tokens=6000,
                                messages=messages,
                                tools=claude_tools if claude_tools else None
                            )
                            
                            # Extract text from the final response
                            for final_content in final_response.content:
                                if final_content.type == "text":
                                    response_text += final_content.text
                                    
                        except Exception as e:
                            error_msg = f"❌ Error executing {tool_name}: {str(e)}"
                            print(error_msg)
                            response_text += f"

{error_msg}"
                
                return response_text
            
        except Exception as e:
            error_msg = f"❌ Error processing query: {str(e)}"
            print(error_msg)
            return error_msg
    
    async def chat_loop(self):
        """
        Main chat loop with user interaction.
        """
        print("
🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")
        print("💬 You can ask questions about GitHub repositories!")
        print("📚 The client can use tools from the FastMCP HTTP server")
        print("🌐 Connected via Server-Sent Events (SSE)")
        print("-" * 60)
        
        while True:
            try:
                # Request user input
                user_input = input("
👤 You: ").strip()
                
                if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
                    print("👋 Bye!")
                    break
                    
                if not user_input:
                    continue
                
                print("
🤔 Claude is thinking...")
                
                # Process query
                response = await self.process_query(user_input)
                
                # Show response
                print(f"
🤖 Claude: {response}")
                
            except KeyboardInterrupt:
                print("

👋 Disconnecting...")
                break
            except Exception as e:
                print(f"
❌ Error in chat: {str(e)}")
                continue
    
    async def cleanup(self):
        """Clean up resources and close connections."""
        print("🧹 Cleaning up resources...")
        # FastMCP Client cleanup is handled automatically by context manager
        await self.exit_stack.aclose()
        print("✅ Resources released")
 
async def main():
    """
    Main function that initializes and runs the FastMCP client.
    """
    # Verify command line arguments
    if len(sys.argv) != 2:
        print("❌ Usage: python client.py &lt;http_server_url&gt;")
        print("📝 Example: python client.py http://localhost:8000")
        print("📝 Note: Now connects to HTTP server instead of executing script")
        sys.exit(1)
    
    server_url = sys.argv[1]
    
    # Validate URL format
    if not server_url.startswith(('http://', 'https://')):
        print("❌ Error: Server URL must start with http:// or https://")
        print("📝 Example: python client.py http://localhost:8000")
        sys.exit(1)
    
    # Create and run client
    client = FastMCPClient()
    
    try:
        # Connect to the server
        await client.connect_to_server(server_url)
        
        # List available tools after connection
        await client.list_available_tools()
        
        # Start chat loop
        await client.chat_loop()
        
    except Exception as e:
        print(f"❌ Fatal error: {str(e)}")
    finally:
        # Ensure resources are cleaned up
        await client.cleanup()
 
if __name__ == "__main__":
    # Entry point of the script
    asyncio.run(main())
	
	
Copied

			

				
Overwriting client_MCP/client.py
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
	
	
Copied

			

				
/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:240: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.
  mcp.mount("sub_mcp", sub_mcp)
DEBUG: Starting FastMCP GitHub server...
DEBUG: Server name: GitHubMCP
[06/28/25 10:33:36] INFO     Starting MCP server 'GitHubMCP' with ]8;id=281189;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.py\server.py]8;;\:]8;id=128713;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1297\1297]8;;\
                             transport 'streamable-http' on
                             http://0.0.0.0:8000/mcp/
INFO:     Started server process [89401]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd client_MCP && source .venv/bin/activate && uv run client.py http://localhost:8000/mcp
	
	
Copied

			

				
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp
✅ Client created successfully
🛠️  Available tools (2):
==================================================
📋 sub_mcp_hello_world
   Description: Returns a simple greeting.
   Parameters:
📋 list_repository_issues
   Description: Lists open issues for a given GitHub repository.
Args:
    owner: The owner of the repository (e.g., 'modelcontextprotocol')
    repo_name: The name of the repository (e.g., 'python-sdk')
Returns:
    list[dict]: A list of dictionaries, each containing information about an issue
   Parameters: owner, repo_name
🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.
💬 You can ask questions about GitHub repositories!
📚 The client can use tools from the FastMCP HTTP server
🌐 Connected via Server-Sent Events (SSE)
------------------------------------------------------------
👤 You:
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
 
# Create FastMCP server
mcp = FastMCP(
    name="GitHubMCP",
    instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
    include_tags={"public"},
    exclude_tags={"first_issue"}
)
 
sub_mcp = FastMCP(
    name="SubMCP",
)
 
@mcp.tool(tags={"public", "production"})
async def list_repository_issues(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 10 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=10"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"private", "development"})
async def list_more_repository_issues(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 100 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=100"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"public", "first_issue"})
async def first_repository_issue(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Gets the first issue ever created in a GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list containing information about the first issue created
    """
    # Get the first issue by sorting by creation date in ascending order
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;per_page=1"
    print(f"Fetching first issue from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No issues found for this repository.")
                return [{"message": "No issues found for this repository."}]
 
            first_issue = issues_data[0]
            
            # Create a detailed summary of the first issue
            summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
            if first_issue.get('comments', 0) &gt; 0:
                summary += f" ({first_issue.get('comments')} comments)"
            
            issue_info = {
                "number": first_issue.get("number"),
                "title": first_issue.get("title"),
                "user": first_issue.get("user", {}).get("login"),
                "url": first_issue.get("html_url"),
                "state": first_issue.get("state"),
                "comments": first_issue.get("comments"),
                "created_at": first_issue.get("created_at"),
                "updated_at": first_issue.get("updated_at"),
                "body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) &gt; 500 else first_issue.get("body", ""),
                "summary": summary
            }
            
            print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")
            
            # Add context information
            result = {
                "repository": f"{owner}/{repo_name}",
                "note": "This is the very first issue created in this repository",
                "first_issue": issue_info
            }
            
            return [result]
            
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; str:
    """
    Returns a simple greeting.
    """
    return "Hello, world!"
 
mcp.mount("sub_mcp", sub_mcp)
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server
    mcp.run(
        transport="stdio"
    )
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github_server.py
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
 
USER_ID = 1234567890
 
# Create FastMCP server
mcp = FastMCP(
    name="GitHubMCP",
    instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
    include_tags={"public"},
    exclude_tags={"first_issue"}
)
 
sub_mcp = FastMCP(
    name="SubMCP",
)
 
@mcp.tool(
    tags={"public", "production"},
    exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM
)
async def list_repository_issues(owner: str, repo_name: str, user_id: int = USER_ID) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 10 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=10"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary,
                "requested_by_user_id": user_id
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"private", "development"})
async def list_more_repository_issues(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Lists open issues for a given GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list of dictionaries, each containing information about an issue
    """
    # Limit to first 100 issues to avoid very long responses
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;per_page=100"
    print(f"Fetching issues from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No open issues found for this repository.")
                return [{"message": "No open issues found for this repository."}]
 
            issues_summary = []
            for issue in issues_data:
                # Create a more concise summary
                summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comments)"
                
                issues_summary.append({
                    "number": issue.get("number"),
                    "title": issue.get("title"),
                    "user": issue.get("user", {}).get("login"),
                    "url": issue.get("html_url"),
                    "comments": issue.get("comments"),
                    "summary": summary
                })
            
            print(f"Found {len(issues_summary)} open issues.")
            
            # Add context information
            result = {
                "total_found": len(issues_summary),
                "repository": f"{owner}/{repo_name}",
                "note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
                "issues": issues_summary
            }
            
            return [result]
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@mcp.tool(tags={"public", "first_issue"})
async def first_repository_issue(owner: str, repo_name: str) -&gt; list[dict]:
    """
    Gets the first issue ever created in a GitHub repository.
 
    Args:
        owner: The owner of the repository (e.g., 'modelcontextprotocol')
        repo_name: The name of the repository (e.g., 'python-sdk')
 
    Returns:
        list[dict]: A list containing information about the first issue created
    """
    # Get the first issue by sorting by creation date in ascending order
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;per_page=1"
    print(f"Fetching first issue from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            issues_data = response.json()
            
            if not issues_data:
                print("No issues found for this repository.")
                return [{"message": "No issues found for this repository."}]
 
            first_issue = issues_data[0]
            
            # Create a detailed summary of the first issue
            summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
            if first_issue.get('comments', 0) &gt; 0:
                summary += f" ({first_issue.get('comments')} comments)"
            
            issue_info = {
                "number": first_issue.get("number"),
                "title": first_issue.get("title"),
                "user": first_issue.get("user", {}).get("login"),
                "url": first_issue.get("html_url"),
                "state": first_issue.get("state"),
                "comments": first_issue.get("comments"),
                "created_at": first_issue.get("created_at"),
                "updated_at": first_issue.get("updated_at"),
                "body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) &gt; 500 else first_issue.get("body", ""),
                "summary": summary
            }
            
            print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")
            
            # Add context information
            result = {
                "repository": f"{owner}/{repo_name}",
                "note": "This is the very first issue created in this repository",
                "first_issue": issue_info
            }
            
            return [result]
            
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 403 and GITHUB_TOKEN:
                error_message += " (Rate limit with token or token lacks permissions?)"
            elif e.response.status_code == 403 and not GITHUB_TOKEN:
                error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"
            
            print(f"GitHub API error: {e.response.status_code}. {error_message}")
            return [{
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message
            }]
        except Exception as e:
            print(f"An unexpected error occurred: {str(e)}")
            return [{"error": f"An unexpected error occurred: {str(e)}"}]
 
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; str:
    """
    Returns a simple greeting.
    """
    return "Hello, world!"
 
mcp.mount("sub_mcp", sub_mcp)
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server
    mcp.run(
        transport="stdio"
    )
	
	
Copied