"""OSINT MCP Server - Unified intelligence gathering."""
import asyncio
import logging
import sys
from typing import Optional
from fastmcp import FastMCP
from osint_mcp.config.settings import get_settings
from osint_mcp.clients import (
CensysClient,
ShodanClient,
HunterClient,
ApolloClient,
LinkedInClient,
PerplexityClient,
HIBPClient,
SherlockClient,
HoleheCl,
DNSClient,
WHOISClient,
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler(sys.stderr)],
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("OSINT Intelligence Server")
# Global clients
_clients: dict = {}
_clients_lock = asyncio.Lock()
async def get_clients() -> dict:
"""Initialize and return all API clients."""
global _clients
if _clients:
return _clients
async with _clients_lock:
if _clients:
return _clients
settings = get_settings()
_clients = {
"censys": CensysClient(api_token=settings.censys_api_token),
"shodan": ShodanClient(api_key=settings.shodan_api_key),
"hunter": HunterClient(api_key=settings.hunter_api_key),
"apollo": ApolloClient(api_key=settings.apollo_api_key),
"linkedin": LinkedInClient(rapidapi_key=settings.rapidapi_key),
"perplexity": PerplexityClient(api_key=settings.perplexity_api_key),
"hibp": HIBPClient(api_key=settings.hibp_api_key),
"sherlock": SherlockClient(),
"holehe": HoleheCl(),
"dns": DNSClient(),
"whois": WHOISClient(),
}
# Start HTTP clients
for name, client in _clients.items():
if hasattr(client, "start"):
try:
await client.start()
except Exception as e:
logger.warning(f"Failed to start {name} client: {e}")
available = settings.get_available_services()
logger.info(f"OSINT MCP initialized. APIs: {[k for k, v in available.items() if v]}")
return _clients
# =============================================================================
# INFRASTRUCTURE INTELLIGENCE
# =============================================================================
@mcp.tool()
async def osint_censys_host(ip: str) -> dict:
"""
Get Censys data for an IP address - services, ports, certificates, ASN.
Args:
ip: IP address to look up
"""
clients = await get_clients()
return await clients["censys"].host_lookup(ip)
@mcp.tool()
async def osint_censys_search(query: str, limit: int = 25) -> dict:
"""
Search Censys for hosts matching a query (e.g., certificates, services).
Args:
query: Censys query language search
limit: Max results (up to 100)
"""
clients = await get_clients()
return await clients["censys"].search(query, limit)
@mcp.tool()
async def osint_shodan_host(ip: str) -> dict:
"""
Get Shodan data for an IP - ports, services, vulnerabilities.
Args:
ip: IP address to look up
"""
clients = await get_clients()
return await clients["shodan"].host_lookup(ip)
@mcp.tool()
async def osint_shodan_domain(domain: str) -> dict:
"""
Get Shodan DNS data - subdomains and DNS records.
Args:
domain: Domain to look up
"""
clients = await get_clients()
return await clients["shodan"].domain_lookup(domain)
@mcp.tool()
async def osint_dns_lookup(domain: str) -> dict:
"""
Perform DNS lookups - A, MX, NS, TXT records.
Args:
domain: Domain to look up
"""
clients = await get_clients()
return await clients["dns"].lookup(domain)
@mcp.tool()
async def osint_whois(domain: str) -> dict:
"""
Get WHOIS registration data for a domain.
Args:
domain: Domain to look up
"""
clients = await get_clients()
return await clients["whois"].lookup(domain)
# =============================================================================
# EMAIL & IDENTITY
# =============================================================================
@mcp.tool()
async def osint_email_search(domain: str, limit: int = 10) -> dict:
"""
Find email addresses for a company domain using Hunter.io.
Args:
domain: Company domain
limit: Max results
"""
clients = await get_clients()
return await clients["hunter"].domain_search(domain, limit)
@mcp.tool()
async def osint_email_finder(domain: str, first_name: str, last_name: str) -> dict:
"""
Find someone's work email address.
Args:
domain: Company domain
first_name: Person's first name
last_name: Person's last name
"""
clients = await get_clients()
return await clients["hunter"].email_finder(domain, first_name, last_name)
@mcp.tool()
async def osint_email_verify(email: str) -> dict:
"""
Verify if an email address is valid and deliverable.
Args:
email: Email to verify
"""
clients = await get_clients()
return await clients["hunter"].email_verify(email)
@mcp.tool()
async def osint_breach_check(email: str) -> dict:
"""
Check if email has been in data breaches (Have I Been Pwned).
Args:
email: Email to check
"""
clients = await get_clients()
return await clients["hibp"].check_breaches(email)
@mcp.tool()
async def osint_email_services(email: str) -> dict:
"""
Detect which services an email is registered with using Holehe.
Args:
email: Email to check
"""
clients = await get_clients()
return await clients["holehe"].check(email)
@mcp.tool()
async def osint_username_search(username: str) -> dict:
"""
Find accounts across platforms for a username using Sherlock.
Args:
username: Username to search
"""
clients = await get_clients()
return await clients["sherlock"].search(username)
# =============================================================================
# PERSON INTELLIGENCE
# =============================================================================
@mcp.tool()
async def osint_person_enrich(
first_name: str,
last_name: str,
organization: Optional[str] = None,
domain: Optional[str] = None,
) -> dict:
"""
Enrich person data using Apollo.io - find email, title, social profiles.
Args:
first_name: Person's first name
last_name: Person's last name
organization: Company name
domain: Company domain
"""
clients = await get_clients()
return await clients["apollo"].find_person(first_name, last_name, organization, domain)
@mcp.tool()
async def osint_linkedin_find(name: str, company: Optional[str] = None) -> dict:
"""
Find someone's LinkedIn profile URL.
Args:
name: Full name
company: Company for disambiguation
"""
clients = await get_clients()
return await clients["linkedin"].find_profile(name, company)
@mcp.tool()
async def osint_linkedin_profile(linkedin_url: str) -> dict:
"""
Get full LinkedIn profile - experience, education, skills.
Args:
linkedin_url: LinkedIn profile URL
"""
clients = await get_clients()
return await clients["linkedin"].get_profile(linkedin_url)
@mcp.tool()
async def osint_person_research(name: str, context: Optional[str] = None) -> dict:
"""
AI-powered person research using Perplexity - synthesizes web info.
Args:
name: Person's name
context: Additional context (company, role, etc.)
"""
clients = await get_clients()
return await clients["perplexity"].research_person(name, context)
# =============================================================================
# COMPANY INTELLIGENCE
# =============================================================================
@mcp.tool()
async def osint_company_enrich(domain: str) -> dict:
"""
Enrich company data using Apollo.io - size, industry, tech stack.
Args:
domain: Company website domain
"""
clients = await get_clients()
return await clients["apollo"].enrich_company(domain)
@mcp.tool()
async def osint_company_people(
organization: Optional[str] = None,
domain: Optional[str] = None,
titles: Optional[list[str]] = None,
limit: int = 10,
) -> dict:
"""
Find people at a company using Apollo.io.
Args:
organization: Company name
domain: Company domain
titles: Filter by titles (e.g., ["CEO", "CTO"])
limit: Max results
"""
clients = await get_clients()
return await clients["apollo"].search_people(organization, domain, titles, limit)
@mcp.tool()
async def osint_linkedin_company(linkedin_url: Optional[str] = None, domain: Optional[str] = None) -> dict:
"""
Get LinkedIn company data.
Args:
linkedin_url: Company LinkedIn URL
domain: Company domain (alternative)
"""
clients = await get_clients()
return await clients["linkedin"].get_company(linkedin_url, domain)
@mcp.tool()
async def osint_company_research(company: str, domain: Optional[str] = None) -> dict:
"""
AI-powered company research using Perplexity.
Args:
company: Company name
domain: Company domain
"""
clients = await get_clients()
return await clients["perplexity"].research_company(company, domain)
# =============================================================================
# UTILITY
# =============================================================================
@mcp.tool()
async def osint_api_status() -> dict:
"""Check which OSINT APIs are configured and available."""
settings = get_settings()
available = settings.get_available_services()
return {
"configured": [k for k, v in available.items() if v],
"missing": [k for k, v in available.items() if not v],
"status": available,
}
@mcp.tool()
async def osint_query(prompt: str) -> dict:
"""
General AI-powered OSINT query using Perplexity.
Args:
prompt: Research question or query
"""
clients = await get_clients()
system = "You are an OSINT researcher. Provide factual, sourced information."
return await clients["perplexity"].query(prompt, system)
# =============================================================================
# SERVER LIFECYCLE
# =============================================================================
async def cleanup():
"""Cleanup on shutdown."""
global _clients
for name, client in _clients.items():
if hasattr(client, "close"):
try:
await client.close()
except Exception as e:
logger.warning(f"Error closing {name}: {e}")
_clients = {}
def main():
"""Entry point."""
import os
transport = os.environ.get("MCP_TRANSPORT", "stdio")
port = int(os.environ.get("MCP_PORT", "5006"))
try:
asyncio.run(get_clients())
logger.info(f"Starting OSINT MCP Server (transport={transport}, port={port})...")
if transport == "sse":
mcp.run(transport="sse", port=port, host="0.0.0.0")
elif transport == "streamable-http":
mcp.run(transport="streamable-http", port=port, host="0.0.0.0")
else:
mcp.run(transport="stdio")
except KeyboardInterrupt:
logger.info("Shutting down...")
finally:
asyncio.run(cleanup())
if __name__ == "__main__":
main()
