MCP with FastMCP (3/4): advanced resources and prompts

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from fastmcp import FastMCP, Context
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, ctx: Context, 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')
        ctx: The context of the request
        user_id: The user ID (automatically injected by the server)
 
    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"
    ctx.info(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:
                ctx.info("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
                })
            
            ctx.info(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.)"
            
            ctx.error(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:
            ctx.error(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)}"}]
 
@mcp.resource("resource://server_info", tags={"public"})
def server_info(ctx: Context) -&gt; str:
    """
    Returns information about the server.
    """
    return {
        "info": "This is the MCP GitHub server development for MaximoFN blog post",
        "requested_id": ctx.request_id
    }
 
@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
			
		

return {
"info": "This is the development of the MCP GitHub server for the MaximoFN blog post",
"requested_id": ctx.request_id
}

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

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

			

				
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:258: 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)
[06/28/25 11:17:41] INFO     Starting MCP server 'GitHubMCP' with transport 'stdio'                          server.py:1246
🛠️  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')
    ctx: The context of the request
    user_id: The user ID (automatically injected by the server)
Returns:
    list[dict]: A list of dictionaries, each containing information about an issue
   Parameters: owner, repo_name
...
🔧 Claude wants to use: read_mcp_resource
📝 Arguments: {'resource_uri': 'resource://server_info'}
📖 Resource read successfully: resource://server_info
🤖 Claude: I'll help you read the server information using the `read_mcp_resource` function. The server information is available at the resource URI "resource://server_info".According to the server information:
- This is the MCP GitHub server development for MaximoFN blog post
- The requested ID is "7"
👤 You: q
👋 Bye!
🧹 Cleaning up resources...
✅ Resources released
			
		

👤 You: Tell me the server info

🤔 Claude is thinking...
🔧 Claude wants to use: read_mcp_resource
📝 Arguments: {'resource_uri': 'resource://server_info'}
📖 Resource read successfully: resource://server_info
🤖 Claude: Te ayudaré a leer la información del servidor usando la función `read_mcp_resource`. La información del servidor está disponible en la URI del recurso "resource://server_info".Según la información del servidor:
- Esta es la publicación del blog de MaximoFN sobre el desarrollo del servidor MCP de GitHub
- The requested ID is "7"

	

		

			< >
			Input
		
		

			Python
			
				
				Copy
			
		
	
	

		
%%writefile gitHub_MCP_server/github_server.py
 
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
import datetime
 
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, ctx: Context, 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')
        ctx: The context of the request
        user_id: The user ID (automatically injected by the server)
 
    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"
    ctx.info(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:
                ctx.info("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
                })
            
            ctx.info(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.)"
            
            ctx.error(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:
            ctx.error(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)}"}]
 
@mcp.resource("resource://server_info", tags={"public"})
def server_info(ctx: Context) -&gt; str:
    """
    Returns information about the server.
    """
    return {
        "info": "This is the MCP GitHub server development for MaximoFN blog post",
        "requested_id": ctx.request_id
    }
 
@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})
async def repository_info(owner: str, repo_name: str, ctx: Context) -&gt; dict:
    """
    Returns detailed information about a 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 context of the request
    
    Returns:
        dict: Repository information including name, description, stats, etc.
    """
    api_url = f"https://api.github.com/repos/{owner}/{repo_name}"
    ctx.info(f"Fetching repository information from {api_url}...")
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, headers=create_github_headers())
            response.raise_for_status()
            repo_data = response.json()
            
            # Extract relevant repository information
            repo_info = {
                "name": repo_data.get("name"),
                "full_name": repo_data.get("full_name"),
                "description": repo_data.get("description"),
                "owner": {
                    "login": repo_data.get("owner", {}).get("login"),
                    "type": repo_data.get("owner", {}).get("type")
                },
                "html_url": repo_data.get("html_url"),
                "clone_url": repo_data.get("clone_url"),
                "ssh_url": repo_data.get("ssh_url"),
                "language": repo_data.get("language"),
                "size": repo_data.get("size"), # Size in KB
                "stargazers_count": repo_data.get("stargazers_count"),
                "watchers_count": repo_data.get("watchers_count"),
                "forks_count": repo_data.get("forks_count"),
                "open_issues_count": repo_data.get("open_issues_count"),
                "default_branch": repo_data.get("default_branch"),
                "created_at": repo_data.get("created_at"),
                "updated_at": repo_data.get("updated_at"),
                "pushed_at": repo_data.get("pushed_at"),
                "is_private": repo_data.get("private"),
                "is_fork": repo_data.get("fork"),
                "is_archived": repo_data.get("archived"),
                "has_issues": repo_data.get("has_issues"),
                "has_projects": repo_data.get("has_projects"),
                "has_wiki": repo_data.get("has_wiki"),
                "license": repo_data.get("license", {}).get("name") if repo_data.get("license") else None,
                "topics": repo_data.get("topics", [])
            }
            
            ctx.info(f"Successfully retrieved information for repository {owner}/{repo_name}")
            return repo_info
            
        except httpx.HTTPStatusError as e:
            error_message = e.response.json().get("message", "No additional message from API.")
            if e.response.status_code == 404:
                error_message = f"Repository {owner}/{repo_name} not found or is private."
            elif 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.)"
            
            ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")
            return {
                "error": f"GitHub API error: {e.response.status_code}",
                "message": error_message,
                "repository": f"{owner}/{repo_name}"
            }
        except Exception as e:
            ctx.error(f"An unexpected error occurred: {str(e)}")
            return {
                "error": f"An unexpected error occurred: {str(e)}",
                "repository": f"{owner}/{repo_name}"
            }
    
 
@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
			
		

@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})
async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:

	

		

			< >
			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 and resources 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 list_available_resources(self):
        """List available resources in the FastMCP server."""
        try:
            # Get list of resources from the server using FastMCP context
            async with self.client as client:
                resources = await client.list_resources()
                
                if resources:
                    print(f"
📚 Available resources ({len(resources)}):")
                    print("=" * 50)
                    
                    for resource in resources:
                        print(f"📄 {resource.uri}")
                        if resource.name:
                            print(f"   Name: {resource.name}")
                        if resource.description:
                            print(f"   Description: {resource.description}")
                        if resource.mimeType:
                            print(f"   MIME Type: {resource.mimeType}")
                        print()
                else:
                    print("⚠️  No resources found in the server")
                    
        except Exception as e:
            print(f"❌ Error listing resources: {str(e)}")
 
    async def read_resource(self, resource_uri: str):
        """
        Read a specific resource from the server.
        
        Args:
            resource_uri: URI of the resource to read
            
        Returns:
            str: Resource content
        """
        try:
            async with self.client as client:
                result = await client.read_resource(resource_uri)
                return result
        except Exception as e:
            print(f"❌ Error reading resource {resource_uri}: {str(e)}")
            return None
 
    async def process_query(self, query: str) -&gt; str:
        """
        Process a user query, interacting with Claude and FastMCP tools and resources.
        
        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 and resources
                tools_list = await client.list_tools()
                resources_list = await client.list_resources()
                
                # 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)
                
                # Add a special tool for reading resources (including template resources)
                resource_description = "Read a resource from the MCP server. "
                if resources_list:
                    # Convert URIs to strings to avoid AnyUrl object issues
                    resource_uris = [str(r.uri) for r in resources_list]
                    resource_description += f"Available static resources: {', '.join(resource_uris)}. "
                
                resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."
                
                claude_tools.append({
                    "name": "read_mcp_resource",
                    "description": resource_description,
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "resource_uri": {
                                "type": "string",
                                "description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"
                            }
                        },
                        "required": ["resource_uri"]
                    }
                })
                
                # 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:
                            if tool_name == "read_mcp_resource":
                                # Handle resource reading
                                resource_uri = tool_args.get("resource_uri")
                                if resource_uri:
                                    tool_result = await client.read_resource(resource_uri)
                                    print(f"📖 Resource read successfully: {resource_uri}")
                                    
                                    # Better handling of resource result
                                    if hasattr(tool_result, 'content'):
                                        # If it's a resource response object, extract content
                                        if hasattr(tool_result.content, 'text'):
                                            result_content = tool_result.content.text
                                        else:
                                            result_content = str(tool_result.content)
                                    else:
                                        # If it's already a string or simple object
                                        result_content = str(tool_result)
                                else:
                                    tool_result = "Error: No resource URI provided"
                                    result_content = tool_result
                            else:
                                # Execute regular tool on the FastMCP server
                                tool_result = await client.call_tool(tool_name, tool_args)
                                print(f"✅ Tool executed successfully")
                                result_content = str(tool_result)
                            
                            # Add tool result to the conversation
                            messages.append({
                                "role": "assistant",
                                "content": response.content
                            })
                            
                            # Format result for Claude
                            if 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 and resources 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 and resources after connection
        await client.list_available_tools()
        await client.list_available_resources()
        
        # 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