"""Sentry MCP server implementation with base tools + instance parameter design."""
import base64
import logging
import os
import re
from typing import Any
from fastmcp import FastMCP
from .client import SentryClientManager
from .config import SentryMCPConfig, SentryRegion, load_config
# Initialize logging
logger = logging.getLogger(__name__)
# Initialize FastMCP
app = FastMCP("sentry", dependencies=["httpx", "pydantic", "pyyaml"])
# Global client manager (initialized on startup)
_client_manager: SentryClientManager | None = None
_config: SentryMCPConfig | None = None
def get_client_manager() -> SentryClientManager:
"""Get the global client manager, initializing if needed."""
global _client_manager, _config
if _client_manager is None:
config_path = os.environ.get("SENTRY_MCP_CONFIG", "/etc/sentry-mcp/config.yaml")
secrets_path = os.environ.get("SENTRY_MCP_SECRETS", "/etc/secrets")
_config = load_config(config_path, secrets_path)
_client_manager = SentryClientManager()
for instance in _config.instances.values():
if instance.auth_token:
_client_manager.register(instance)
logger.info(f"Registered Sentry instance: {instance.name} ({instance.region})")
else:
logger.warning(f"Skipping instance {instance.name}: no auth token")
return _client_manager
def get_config() -> SentryMCPConfig:
"""Get the global config, initializing if needed."""
get_client_manager() # Ensures config is loaded
if _config is None:
raise RuntimeError("Config not initialized")
return _config
def _parse_issue_id_or_url(issue_id_or_url: str) -> str:
"""Extract issue ID from URL or return as-is.
Supports:
- Direct ID: "12345" or "PROJ-123"
- URL: "https://sentry.io/organizations/org/issues/12345/"
"""
# Check if it's a URL
url_pattern = r"issues/(\d+)"
match = re.search(url_pattern, issue_id_or_url)
if match:
return match.group(1)
return issue_id_or_url
# ===== DISCOVERY TOOL =====
@app.tool()
async def list_sentry_instances() -> dict[str, Any]:
"""List all configured Sentry instances and their metadata.
CALL THIS FIRST to discover available Sentry accounts before using other tools.
Each instance represents a separate Sentry organization with its own data.
Returns:
Dictionary with:
- instances: List of available Sentry instances with metadata
- note: Usage guidance for other tools
Example response:
{
"instances": [
{
"name": "sentryUS",
"region": "US",
"api_url": "sentry.io",
"org_slug": "acme-corp",
"data_residency": "US",
"description": "US customer data and services"
},
{
"name": "sentryEU",
"region": "EU",
"api_url": "de.sentry.io",
"org_slug": "acme-corp-eu",
"data_residency": "EU",
"description": "EU customer data (GDPR)"
}
],
"note": "Use instance name when calling other Sentry tools..."
}
"""
config = get_config()
manager = get_client_manager()
available_instances = manager.list_instances()
instances = []
for name in available_instances:
instance_config = config.instances.get(name)
if instance_config:
# Build region display
region_display = instance_config.region.value.upper()
if instance_config.region == SentryRegion.CUSTOM:
region_display = "Self-hosted"
# Build API URL display
api_url = instance_config.api_base_url.replace("https://", "")
instances.append({
"name": name,
"region": region_display,
"api_url": api_url,
"org_slug": instance_config.org_slug,
"data_residency": instance_config.labels.data_residency,
"description": instance_config.labels.description,
"team": instance_config.labels.team,
"environment": instance_config.labels.environment,
})
return {
"instances": instances,
"note": (
"Use instance name when calling other Sentry tools. "
"Query both instances if unsure which region contains the data."
),
}
# ===== ORGANIZATION & PROJECT DISCOVERY =====
@app.tool()
async def find_organizations(
instance: str,
) -> dict[str, Any]:
"""Find organizations accessible in a Sentry instance.
Note: Internal Integration tokens typically only have access to their own org.
Args:
instance: Sentry instance name (from list_sentry_instances)
Returns:
Dictionary with list of organizations
"""
manager = get_client_manager()
client = manager.get(instance)
orgs = await client.list_organizations()
return {
"instance": instance,
"organizations": orgs,
"count": len(orgs),
}
@app.tool()
async def find_teams(
instance: str,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Find teams in a Sentry organization.
Args:
instance: Sentry instance name (from list_sentry_instances)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with list of teams
"""
manager = get_client_manager()
client = manager.get(instance)
teams = await client.list_teams(org_slug)
return {
"instance": instance,
"org_slug": org_slug or client.config.org_slug,
"teams": teams,
"count": len(teams),
}
@app.tool()
async def find_projects(
instance: str,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Find projects in a Sentry organization.
Use this to discover available projects before searching for issues.
Args:
instance: Sentry instance name (from list_sentry_instances)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with list of projects including slug, name, platform, etc.
"""
manager = get_client_manager()
client = manager.get(instance)
projects = await client.list_projects(org_slug)
return {
"instance": instance,
"org_slug": org_slug or client.config.org_slug,
"projects": projects,
"count": len(projects),
}
@app.tool()
async def find_releases(
instance: str,
project_slug: str | None = None,
query: str | None = None,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Find releases in a Sentry organization or project.
Args:
instance: Sentry instance name (from list_sentry_instances)
project_slug: Filter by project (optional)
query: Search query for releases (optional)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with list of releases
"""
manager = get_client_manager()
client = manager.get(instance)
releases = await client.list_releases(org_slug, project_slug, query)
return {
"instance": instance,
"org_slug": org_slug or client.config.org_slug,
"project_slug": project_slug,
"releases": releases,
"count": len(releases),
}
# ===== ISSUE INVESTIGATION =====
@app.tool()
async def get_issue_details(
instance: str,
issue_id: str,
) -> dict[str, Any]:
"""Get detailed information about a specific Sentry issue.
Retrieves comprehensive details including error message, stack trace,
affected users, tags, and more.
Args:
instance: Sentry instance name (from list_sentry_instances)
issue_id: Issue ID (numeric) or full Sentry issue URL
Returns:
Dictionary with full issue details
Examples:
- get_issue_details(instance="sentryUS", issue_id="12345")
- get_issue_details(instance="sentryEU", issue_id="https://sentry.io/organizations/acme/issues/12345/")
"""
manager = get_client_manager()
client = manager.get(instance)
# Parse issue ID from URL if needed
parsed_id = _parse_issue_id_or_url(issue_id)
issue = await client.get_issue(parsed_id)
return {
"instance": instance,
"issue": issue,
}
@app.tool()
async def search_issues(
instance: str,
project_slug: str,
query: str = "is:unresolved",
stats_period: str = "24h",
limit: int = 25,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Search for grouped issues in a Sentry project.
Returns a list of issue groups (not individual events). Use this to find
error patterns, unresolved issues, or issues matching specific criteria.
FIRST call find_projects() to discover available projects.
Args:
instance: Sentry instance name (from list_sentry_instances)
project_slug: Project slug to search in (from find_projects)
query: Sentry search query. Examples:
- "is:unresolved" (default) - unresolved issues
- "is:unresolved level:error" - unresolved errors
- "is:unresolved assigned:me" - assigned to token owner
- "is:unresolved firstSeen:-24h" - new in last 24h
- "is:unresolved browser.name:Chrome" - Chrome-specific
- "payment" - issues containing "payment"
stats_period: Time period for stats (e.g., "24h", "14d", "30d")
limit: Maximum number of issues to return (default 25)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with issues list and count
"""
manager = get_client_manager()
client = manager.get(instance)
result = await client.search_issues(
project_slug=project_slug,
org_slug=org_slug,
query=query,
stats_period=stats_period,
limit=limit,
)
return {
"instance": instance,
"project_slug": project_slug,
"query": query,
**result,
}
@app.tool()
async def search_issue_events(
instance: str,
issue_id: str,
full: bool = True,
limit: int = 25,
) -> dict[str, Any]:
"""Search and filter events within a specific issue.
Use this to see individual occurrences of an issue, including
stack traces, request data, and user context for each event.
Args:
instance: Sentry instance name (from list_sentry_instances)
issue_id: Issue ID (numeric) or full Sentry issue URL
full: Include full event details (default True)
limit: Maximum number of events to return (default 25)
Returns:
Dictionary with events list
"""
manager = get_client_manager()
client = manager.get(instance)
parsed_id = _parse_issue_id_or_url(issue_id)
result = await client.search_issue_events(
issue_id=parsed_id,
full=full,
limit=limit,
)
return {
"instance": instance,
"issue_id": parsed_id,
**result,
}
# ===== EVENT & TRACE ANALYSIS =====
@app.tool()
async def search_events(
instance: str,
query: str | None = None,
project: str | None = None,
field: str | None = None,
stats_period: str = "24h",
per_page: int = 50,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Search events and perform counts/aggregations.
USE THIS for statistics and counting questions like:
- "How many errors in the last 24 hours?"
- "What are the top error types?"
- "Error count by project"
Args:
instance: Sentry instance name (from list_sentry_instances)
query: Sentry search query (e.g., "level:error", "transaction:/api/users")
project: Project slug to filter by (optional)
field: Comma-separated fields to return. Examples:
- "count()" - total count
- "count(),title" - count with titles
- "count(),project" - count by project
- "count_unique(user)" - unique users affected
- "p50(transaction.duration),p95(transaction.duration)" - percentiles
stats_period: Time period (e.g., "24h", "7d", "14d", "30d")
per_page: Results per page (default 50)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with event data and aggregations
"""
manager = get_client_manager()
client = manager.get(instance)
# Parse field string to list
field_list = [f.strip() for f in field.split(",")] if field else None
result = await client.search_events(
org_slug=org_slug,
project=project,
query=query,
field=field_list,
stats_period=stats_period,
per_page=per_page,
)
return {
"instance": instance,
"query": query,
"stats_period": stats_period,
**result,
}
@app.tool()
async def get_trace_details(
instance: str,
trace_id: str,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Get trace overview and span breakdown by trace ID.
Use this to analyze distributed traces, see span timing,
and understand request flow across services.
Args:
instance: Sentry instance name (from list_sentry_instances)
trace_id: Trace ID (32-character hex string)
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with trace details and spans
"""
manager = get_client_manager()
client = manager.get(instance)
result = await client.get_trace(trace_id, org_slug)
return {
"instance": instance,
"trace_id": trace_id,
**result,
}
@app.tool()
async def get_event_attachment(
instance: str,
project_slug: str,
event_id: str,
attachment_id: str,
org_slug: str | None = None,
) -> dict[str, Any]:
"""Download an attachment from a Sentry event.
Attachments can include screenshots, log files, or other debug data
uploaded with the event.
Args:
instance: Sentry instance name (from list_sentry_instances)
project_slug: Project slug
event_id: Event ID
attachment_id: Attachment ID
org_slug: Organization slug (optional, defaults to instance's configured org)
Returns:
Dictionary with attachment content (base64 encoded) and metadata
"""
manager = get_client_manager()
client = manager.get(instance)
content = await client.get_event_attachment(
org_slug=org_slug,
project_slug=project_slug,
event_id=event_id,
attachment_id=attachment_id,
)
# Encode binary content as base64 for JSON transport
encoded = base64.b64encode(content).decode("utf-8")
return {
"instance": instance,
"project_slug": project_slug,
"event_id": event_id,
"attachment_id": attachment_id,
"content_base64": encoded,
"size_bytes": len(content),
}
