MCP with ThinAgents

# server.py
import json
import requests
from fastmcp import FastMCP, Context


def get_bearer_token(ctx: Context) -> str:
    """Extract bearer token from Authorization header."""
    request = ctx.get_http_request()
    headers = request.headers
    authorization_header = headers.get('Authorization')

    if not authorization_header:
        raise ValueError("Authorization header missing")

    parts = authorization_header.split()
    if len(parts) != 2 or parts[0] != 'Bearer':
        raise ValueError("Invalid Authorization header format")

    return parts[1]


# Create MCP server
mcp = FastMCP("Weather_Service", streamable_http_path='/')

@mcp.tool()
def fetch_weather(city: str, ctx: Context) -> dict:
    """
    Fetch current weather data for a given city using OpenWeather API.
    Requires Authorization: Bearer <openweather_api_key> header.
    """
    try:
        api_key = get_bearer_token(ctx)
    except ValueError as e:
        return {
            "status": "Failed",
            "error": f"Authorization error: {str(e)}"
        }

    base_url = "http://api.openweathermap.org/data/2.5/weather"

    params = {
        "q": city,
        "appid": api_key,
        "units": "metric"
    }

    try:
        response = requests.get(base_url, params=params)
        response.raise_for_status()
        weather_data = response.json()

        with open("weather_data.json", "w") as file:
            json.dump(weather_data, file, indent=4)

        return {
            "temperature": weather_data["main"]["temp"],
            "feels_like": weather_data["main"]["feels_like"],
            "humidity": weather_data["main"]["humidity"],
            "description": weather_data["weather"][0]["description"],
            "wind_speed": weather_data["wind"]["speed"]
        }

    except requests.exceptions.RequestException as e:
        return {"error": f"Failed to fetch weather data: {str(e)}"}
    except (KeyError, ValueError) as e:
        return {"error": f"Failed to parse weather data: {str(e)}"}


if __name__ == "__main__":
    mcp.run(transport="sse", port=8000, host="localhost")

import os
import requests
from fastmcp import FastMCP, Context


mcp = FastMCP("GitHub HTTP Server", streamable_http_path='/')

def get_bearer_token(ctx: Context) -> str:
    """Extract bearer token from Authorization header."""
    request = ctx.get_http_request()
    headers = request.headers
    authorization_header = headers.get('Authorization')
    
    if not authorization_header:
        raise ValueError("Authorization header missing")
    
    parts = authorization_header.split()
    if len(parts) != 2 or parts[0] != 'Bearer':
        raise ValueError("Invalid Authorization header format")
    
    return parts[1]


def _github_headers(token: str) -> dict:
    headers = {"Accept": "application/vnd.github+json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


@mcp.tool()
def list_repos(user: str, ctx: Context) -> list[str]:
    """List public repositories for a GitHub user/org. Requires Authorization: Bearer <github_token> header."""
    try:
        token = get_bearer_token(ctx)
    except ValueError as e:
        return [f"Authorization error: {str(e)}"]
    
    url = f"https://api.github.com/users/{user}/repos?per_page=100"
    resp = requests.get(url, headers=_github_headers(token), timeout=20)
    resp.raise_for_status()
    return [r["full_name"] for r in resp.json()]


@mcp.tool()
def repo_readme(repo: str, ctx: Context, ref: str = "") -> str:
    """Fetch README.md for repo ('owner/name'). Optionally specify ref (branch/tag/sha). Requires Authorization: Bearer <github_token> header."""
    try:
        token = get_bearer_token(ctx)
    except ValueError as e:
        return f"Authorization error: {str(e)}"
    
    owner, name = repo.split("/", 1)
    url = f"https://api.github.com/repos/{owner}/{name}/readme"
    if ref:
        url += f"?ref={ref}"
    resp = requests.get(url, headers={**_github_headers(token), "Accept": "application/vnd.github.raw"}, timeout=20)
    if resp.status_code == 404:
        return "README not found."
    resp.raise_for_status()
    return resp.text


@mcp.tool()
def repo_file(repo: str, path: str, ctx: Context, ref: str = "") -> str:
    """Fetch raw file content from a repo at given path. Optionally specify ref. Requires Authorization: Bearer <github_token> header."""
    try:
        token = get_bearer_token(ctx)
    except ValueError as e:
        return f"Authorization error: {str(e)}"
    
    owner, name = repo.split("/", 1)
    base = f"https://api.github.com/repos/{owner}/{name}/contents/{path}"
    url = f"{base}?ref={ref}" if ref else base
    resp = requests.get(url, headers={**_github_headers(token), "Accept": "application/vnd.github.raw"}, timeout=20)
    if resp.status_code == 404:
        return "File not found."
    resp.raise_for_status()
    return resp.text


if __name__ == "__main__":
    host = os.getenv("MCP_HOST", "127.0.0.1")
    port = int(os.getenv("MCP_PORT", "8103"))
    mcp.run(transport="streamable-http", host=host, port=port)