MCP con FastMCP (1/4): crea tu primer servidor y gestiona sus tools

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!mkdir gitHub_MCP_server
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd gitHub_MCP_server && uv init .
	
	
Copied

			

				
Initialized project `github-mcp-server` at `/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server`
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd gitHub_MCP_server && uv venv
	
	
Copied

			

				
Using CPython 3.11.11
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd gitHub_MCP_server && uv add anthropic fastmcp python-dotenv requests
	
	
Copied

			

				
Resolved 42 packages in 34ms
Installed 40 packages in 71ms
+ annotated-types==0.7.0
+ anyio==4.9.0
+ authlib==1.6.0
+ certifi==2025.6.15
+ cffi==1.17.1
+ charset-normalizer==3.4.2
+ click==8.2.1
+ cryptography==45.0.4
+ distro==1.9.0
+ exceptiongroup==1.3.0
+ fastmcp==2.9.0
+ h11==0.16.0
+ httpcore==1.0.9
+ httpx==0.28.1
+ httpx-sse==0.4.0
+ idna==3.10
+ jiter==0.10.0
+ markdown-it-py==3.0.0
+ mcp==1.9.4
+ mdurl==0.1.2
+ openapi-pydantic==0.5.1
+ pycparser==2.22
+ pydantic==2.11.7
+ pydantic-core==2.33.2
+ pydantic-settings==2.10.0
+ pygments==2.19.2
+ python-dotenv==1.1.1
+ python-multipart==0.0.20
+ requests==2.32.4
+ rich==14.0.0
+ shellingham==1.5.4
+ sniffio==1.3.1
+ sse-starlette==2.3.6
+ starlette==0.47.1
+ typer==0.16.0
+ typing-extensions==4.14.0
+ typing-inspection==0.4.1
+ typing-inspection==0.4.1
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!mkdir client_MCP
	
	
Copied

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd client_MCP && uv init .
	
	
Copied

			

				
Initialized project `client-mcp` at `/Users/macm1/Documents/web/portafolio/posts/client_MCP`
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd client_MCP && uv venv
	
	
Copied

			

				
Using CPython 3.11.11
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
!cd client_MCP && uv add anthropic fastmcp python-dotenv requests
	
	
Copied

			

				
Resolved 42 packages in 307ms
Prepared 5 packages in 115ms
Installed 40 packages in 117ms
+ annotated-types==0.7.0
+ anthropic==0.55.0
+ anyio==4.9.0
+ authlib==1.6.0
+ certifi==2025.6.15
+ cffi==1.17.1
+ charset-normalizer==3.4.2
+ click==8.2.1
+ cryptography==45.0.4
+ distro==1.9.0
+ exceptiongroup==1.3.0
+ fastmcp==2.9.0
+ h11==0.16.0
+ httpcore==1.0.9
+ httpx==0.28.1
+ httpx-sse==0.4.0
+ idna==3.10
...
+ requests==2.32.4
+ rich==14.0.0
+ shellingham==1.5.4
+ sniffio==1.3.1
+ sse-starlette==2.3.6
+ starlette==0.47.1
+ typer==0.16.0
+ typing-extensions==4.14.0
+ typing-inspection==0.4.1
+ typing-inspection==0.4.1
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile client_MCP/.env
 
ANTHROPIC_API_KEY="ANTHROPIC_API_KEY"
	
	
Copied

			

				
Writing client_MCP/.env
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
from mcp.server.fastmcp import FastMCP
 
# Create an MCP server
mcp = FastMCP("GitHubMCP")
 
if __name__ == "__main__":
    # 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.py
 
import os
from dotenv import load_dotenv
 
# Load the GitHub token from the .env file
load_dotenv()
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
 
# Check if the GitHub token is configured
if not GITHUB_TOKEN:
    print("WARNING: The GITHUB_TOKEN environment variable is not configured.")
    print("Requests to the GitHub API may fail due to rate limits.")
    print("Create a .env file in this directory with GITHUB_TOKEN='your_token_here'")
    raise ValueError("GITHUB_TOKEN is not configured")
 
# Helper function to create headers for GitHub API requests
def create_github_headers():
    headers = {}
    if GITHUB_TOKEN:
        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
    # GitHub recommends including a User-Agent
    headers["User-Agent"] = "MCP_GitHub_Server_Example"
    headers["Accept"] = "application/vnd.github.v3+json" # Good practice
    return headers
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github.py
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/.env
 
GITHUB_TOKEN = "GITHUB_TOKEN"
	
	
Copied

			

				
Overwriting gitHub_MCP_server/.env
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
 
# Create a FastMCP server
mcp = FastMCP("GitHubMCP")
 
@mcp.tool()
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 the first 10 issues to avoid 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', 'Sin título')}"
                if issue.get('comments', 0) &gt; 0:
                    summary += f" ({issue.get('comments')} comentarios)"
                
                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": "Mostrando los primeros 10 issues abiertos" if len(issues_summary) == 10 else f"Mostrando todos los {len(issues_summary)} issues abiertos",
                "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)}"}]
 
if __name__ == "__main__":
    print("DEBUG: Starting GitHub FastMCP server...")
    print(f"DEBUG: Server name: {mcp.name}")
    print("DEBUG: Available tools: list_repository_issues")
    
    # Initialize and run the server
    mcp.run()
	
	
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_script_path: str):
        """
        Connect to the specified FastMCP server.
        
        Args:
            server_script_path: Path to the server script (Python)
        """
        print(f"🔗 Connecting to FastMCP server: {server_script_path}")
        
        # Determine the server type based on the extension
        if not server_script_path.endswith('.py'):
            raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")
        
        # Create FastMCP client 
        self.client = Client(server_script_path)
        # Note: FastMCP Client automatically infers transport from .py files
        
        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 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 server")
        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;path_to_fastmcp_server&gt;")
        print("📝 Example: python client.py ../MCP_github/github_server.py")
        sys.exit(1)
    
    server_script_path = sys.argv[1]
    
    # Create and run client
    client = FastMCPClient()
    
    try:
        # Connect to the server
        await client.connect_to_server(server_script_path)
        
        # 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 client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
	
	
Copied

			

				
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
[06/28/25 09:22:09] INFO     Starting MCP server 'GitHubMCP' with transport 'stdio'                          server.py:1246
🛠️  Available tools (1):
==================================================
📋 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 client started. Write 'quit', 'q', 'exit', 'salir' to exit.
💬 You can ask questions about GitHub repositories!
📚 The client can use tools from the FastMCP server
------------------------------------------------------------
👤 You: Tell me de issues of repository transformers of huggingface
🤔 Claude is thinking...
🔧 Claude wants to use: list_repository_issues
📝 Arguments: {'owner': 'huggingface', 'repo_name': 'transformers'}
✅ Tool executed successfully
🤖 Claude: I'll help you list the issues from the Hugging Face transformers repository. Let me use the `list_repository_issues` function with the appropriate parameters.I'll summarize the current open issues from the Hugging Face transformers repository. Here are the 10 most recent open issues:
1. [#39097] Core issue about saving models with multiple shared tensor groups when dispatched
2. [#39096] Pull request to fix position index in v4.52.4
3. [#39095] Issue with Qwen2_5_VLVisionAttention flash attention missing 'is_causal' attribute
4. [#39094] Documentation improvement for PyTorch examples
5. [#39093] Style change PR for lru_cache decorator
6. [#39091] Compatibility issue with sentencepiece on Windows in Python 3.13
7. [#39090] Pull request for fixing bugs in finetune and batch inference
8. [#39089] Bug report for LlavaOnevisonConfig initialization in version 4.52.4
9. [#39087] Documentation PR for Gemma 3n audio encoder
10. [#39084] Pull request for refactoring gemma3n
Note that this is showing the 10 most recent open issues, and there might be more issues in the repository. Each issue has a link where you can find more details about the specific problem or proposed changes.
Would you like more specific information about any of these issues?
👤 You: q
👋 Bye!
🧹 Cleaning up resources...
✅ Resources released
			
		

🛠️  Available tools (1):
==================================================
📋 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

👤 You: Tell me de issues of repository transformers of huggingface

🤔 Claude is thinking...
🔧 Calling tool: list_repository_issues
📝 Arguments: {'owner': 'huggingface', 'repo_name': 'transformers'}
✅ Tool executed successfully

🤖 Claude: I'll help you list the issues from the Hugging Face transformers repository. Let me use the `list_repository_issues` function with the appropriate parameters.I'll summarize the current open issues from the Hugging Face transformers repository. Here are the 10 most recent open issues:

1. [#39097] Core issue about saving models with multiple shared tensor groups when dispatched
2. [#39096] Pull request to fix position index in v4.52.4
3. [#39095] Issue with Qwen2_5_VLVisionAttention flash attention missing 'is_causal' attribute
4. [#39094] Documentation improvement for PyTorch examples
5. [#39093] Style change PR for lru_cache decorator
6. [#39091] Compatibility issue with sentencepiece on Windows in Python 3.13
7. [#39090] Pull request for fixing bugs in finetune and batch inference
8. [#39089] Bug report for LlavaOnevisonConfig initialization in version 4.52.4
9. [#39087] Documentation PR for Gemma 3n audio encoder
10. [#39084] Pull request for refactoring gemma3n

Note that this is showing the 10 most recent open issues, and there might be more issues in the repository. Each issue has a link where you can find more details about the specific problem or proposed changes.

Would you like more specific information about any of these issues?

mcp = FastMCP(
  name="GitHubMCP",
  instructions="""
  This server provides tools, resources and prompts to interact with the GitHub API.
  """
)

	

		

			< >
			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.
    """
)
 
@mcp.tool()
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')
        ctx: The MCP context for logging.
 
    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)}"}]
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server
    mcp.run()
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github_server.py
			
		

mcp = FastMCP(
  name="GitHubMCP",
  instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
  include_tags={"public"}
)

@mcp.tool(tags=production)

	

		

			< >
			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"}
)
 
@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)}"}]
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server
    mcp.run()
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github_server.py
			
		

mcp = FastMCP(
  name="GitHubMCP",
  instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
  include_tags={"public"}
)

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

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

			

				
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
[06/28/25 09:44:55] INFO     Starting MCP server 'GitHubMCP' with ]8;id=896921;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py\server.py]8;;\:]8;id=507812;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1246\1246]8;;\
                             transport 'stdio'
🛠️  Available tools (1):
==================================================
📋 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')
    ctx: The MCP context for logging.
Returns:
    list[dict]: A list of dictionaries, each containing information about an issue
   Parameters: owner, repo_name
🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.
💬 You can ask questions about GitHub repositories!
📚 The client can use tools from the FastMCP server
------------------------------------------------------------
👤 You:
			
		

🛠️  Available tools (1):
==================================================
📋 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')
  ctx: The MCP context for logging.

Returns:
  list[dict]: A list of dictionaries, each containing information about an issue
 Parameters: owner, repo_name

	

		

			< >
			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={"private"}
)
 
@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)}"}]
 
if __name__ == "__main__":
    print("DEBUG: Starting FastMCP GitHub server...")
    print(f"DEBUG: Server name: {mcp.name}")
    
    # Initialize and run the server
    mcp.run()
	
	
Copied

			

				
Overwriting gitHub_MCP_server/github_server.py
			
		

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

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

			

				
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
[06/28/25 09:51:48] INFO     Starting MCP server 'GitHubMCP' with ]8;id=921531;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py\server.py]8;;\:]8;id=418078;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1246\1246]8;;\
                             transport 'stdio'
🛠️  Available tools (1):
==================================================
📋 list_more_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 client started. Write 'quit', 'q', 'exit', 'salir' to exit.
💬 You can ask questions about GitHub repositories!
📚 The client can use tools from the FastMCP server
------------------------------------------------------------
👤 You: