web3-jobs-mcp

import os import httpx import re from typing import List, Dict, Any, Optional from mcp.server.fastmcp import FastMCP, Context from mcp.server.fastmcp.prompts import base from dotenv import load_dotenv from email.utils import parsedate_to_datetime from datetime import datetime # Load environment variables load_dotenv() API_TOKEN = os.getenv("WEB3_CAREER_API_TOKEN") if not API_TOKEN: raise ValueError("WEB3_CAREER_API_TOKEN environment variable is not set") # Initialize MCP server mcp = FastMCP("Web3 Jobs Server", dependencies=["httpx", "python-dotenv"]) # Base URL for the API BASE_URL = "https://web3.career/api/v1" async def fetch_jobs(params: Dict[str, Any] = None) -> List[Dict]: """Fetch jobs from web3.career API with given parameters.""" async with httpx.AsyncClient() as client: try: params = params or {} params["token"] = API_TOKEN response = await client.get(BASE_URL, params=params) response.raise_for_status() # API returns a list with metadata and job listings data = response.json() if len(data) > 2: return data[2] # Job listings are in the third element return [] except httpx.HTTPStatusError as e: raise ValueError(f"API request failed: {e}") except Exception as e: raise ValueError(f"Error fetching jobs: {e}") def strip_html(text: str) -> str: """Remove HTML tags from text and clean up whitespace.""" if not text: return "" # Remove HTML tags clean = re.sub(r'<[^>]+>', '', text) # Replace multiple whitespace with single space, strip newlines clean = re.sub(r'\s+', ' ', clean).strip() return clean def jobs_to_markdown(jobs: List[Dict], show_description: bool) -> str: """Convert job list to a Markdown list.""" lines = ["# Web3 Job Listings\n"] for job in jobs: description = strip_html(job.get("description", "")) if show_description else "" # Truncate description to 100 characters for readability if description: description = description[:100] + ("..." if len(description) > 100 else "") apply_url = f"[Apply]({job['apply_url']})" # Get publication date from 'date_epoch' dt = datetime.fromtimestamp(int(job["date_epoch"])) published_at = dt.strftime("%Y-%m-%d") lines.append(f"- **Job ID**: {job['id']}") lines.append(f" - **Title**: {job['title']}") lines.append(f" - **Company**: {job['company']}") lines.append(f" - **Location**: {job['location']}") lines.append(f" - **Remote**: {'Yes' if job['is_remote'] else 'No'}") lines.append(f" - **Published At**: {published_at}") lines.append(f" - **Apply URL**: {apply_url}") if show_description: lines.append(f" - **Description**: {description}") lines.append("") # Blank line between jobs lines.append("*Source: web3.career*") return "\n".join(lines) # Tool: Query jobs with filters @mcp.tool() async def query_jobs( remote: Optional[bool] = None, country: Optional[str] = None, tag: Optional[str] = None, limit: Optional[int] = 50, ctx: Context = None ) -> str: """ Query Web3 job listings from the web3.career API with optional filters. Returns a Markdown-formatted list of jobs, including Job ID, Title, Company, Location, Remote status, Published At date, Apply URL, and Description (always included). Parameters: remote (Optional[bool]): Filter for remote jobs. Set to True for remote-only, False for non-remote, or None for no filter. Defaults to None. country (Optional[str]): Filter by country (e.g., "United States", "Canada"). Case-insensitive, spaces are converted to hyphens (e.g., "united-states"). Defaults to None (no country filter). tag (Optional[str]): Filter by job tag (e.g., "react", "blockchain"). Case-insensitive. Defaults to None (no tag filter). limit (Optional[int]): Maximum number of jobs to return. Must be between 1 and 100. Defaults to 50. Raises ValueError if limit exceeds 100. ctx (Context): MCP context for logging and internal use. Automatically provided by the MCP framework. Defaults to None. Returns: str: A Markdown string containing a list of job listings with the specified fields. Notes: - Descriptions are always included (show_description is hardcoded to True), truncated to 100 characters, and have HTML tags removed. - Published At is derived from the 'date_epoch' field, formatted as YYYY-MM-DD. - Apply URL is included as a clickable Markdown link for each job. - Source is credited to web3.career per API terms. """ if limit > 100: raise ValueError("Limit cannot exceed 100") params = {"limit": limit} if remote is not None: params["remote"] = str(remote).lower() if country: params["country"] = country.lower().replace(" ", "-") if tag: params["tag"] = tag.lower() ctx.info(f"Querying jobs with parameters: {params}") jobs = await fetch_jobs(params) ctx.info(f"Retrieved {len(jobs)} jobs") return jobs_to_markdown(jobs, show_description=True) # Prompt: Generate a job search query @mcp.prompt() def search_jobs_prompt(role: str, location: Optional[str] = None) -> List[base.Message]: """Generate a prompt for searching Web3 jobs based on role and optional location.""" query = f"Find Web3 jobs for a {role} role" if location: query += f" in {location}" query += ". Provide job titles, companies, locations, and application links from web3.career." return [ base.UserMessage(query), base.AssistantMessage("I'll search for relevant Web3 jobs. Here are the results:") ] if __name__ == "__main__": mcp.run()