"""STDIO transport server for MCP tool search.
This module provides the STDIO transport for the MCP server,
suitable for local development and one-off Heroku dyno execution.
Usage:
python -m src.stdio_server
"""
import asyncio
import logging
import sys
import traceback
from mcp.server.stdio import stdio_server
from .tools import mcp_server
from .catalog import catalog, ToolDefinition
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr, # Log to stderr to not interfere with STDIO
)
logger = logging.getLogger(__name__)
def load_sample_tools():
"""Load sample tools for demonstration purposes."""
sample_tools = [
ToolDefinition(
name="get_weather",
description="Get the current weather conditions for a specific location",
input_schema={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state/country, e.g. San Francisco, CA"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Temperature unit"
}
},
"required": ["location"]
},
tags=["weather", "location"]
),
ToolDefinition(
name="get_forecast",
description="Get weather forecast for the next several days",
input_schema={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The location for the forecast"
},
"days": {
"type": "integer",
"description": "Number of days to forecast",
"minimum": 1,
"maximum": 14
}
},
"required": ["location"]
},
tags=["weather", "forecast"]
),
ToolDefinition(
name="search_files",
description="Search through files in a workspace using keywords or patterns",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query or pattern"
},
"file_types": {
"type": "array",
"items": {"type": "string"},
"description": "File extensions to search (e.g., ['.py', '.js'])"
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return"
}
},
"required": ["query"]
},
tags=["files", "search"]
),
ToolDefinition(
name="send_email",
description="Send an email message to one or more recipients",
input_schema={
"type": "object",
"properties": {
"to": {
"type": "array",
"items": {"type": "string"},
"description": "Email recipients"
},
"subject": {
"type": "string",
"description": "Email subject line"
},
"body": {
"type": "string",
"description": "Email body content"
}
},
"required": ["to", "subject", "body"]
},
tags=["email", "communication"]
),
ToolDefinition(
name="query_database",
description="Execute a SQL query against the database",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL query to execute"
},
"database": {
"type": "string",
"description": "Database name to query"
}
},
"required": ["query"]
},
tags=["database", "sql"]
),
ToolDefinition(
name="create_calendar_event",
description="Create a new event in the calendar",
input_schema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Event title"
},
"start_time": {
"type": "string",
"description": "Start time in ISO format"
},
"end_time": {
"type": "string",
"description": "End time in ISO format"
},
"attendees": {
"type": "array",
"items": {"type": "string"},
"description": "Email addresses of attendees"
}
},
"required": ["title", "start_time", "end_time"]
},
tags=["calendar", "scheduling"]
),
ToolDefinition(
name="get_stock_price",
description="Get the current stock price for a given ticker symbol",
input_schema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock ticker symbol (e.g., AAPL, GOOGL)"
}
},
"required": ["symbol"]
},
tags=["finance", "stocks"]
),
ToolDefinition(
name="translate_text",
description="Translate text from one language to another",
input_schema={
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Text to translate"
},
"source_language": {
"type": "string",
"description": "Source language code (e.g., en, es, fr)"
},
"target_language": {
"type": "string",
"description": "Target language code"
}
},
"required": ["text", "target_language"]
},
tags=["language", "translation"]
),
]
catalog.register_tools(sample_tools)
logger.info(f"Loaded {len(sample_tools)} sample tools")
async def main():
"""Run the STDIO MCP server."""
logger.info("Starting Tool Search MCP Server (STDIO mode)")
# Load sample tools if catalog is empty
if catalog.count() == 0:
load_sample_tools()
try:
async with stdio_server() as (read_stream, write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options()
)
except asyncio.CancelledError:
logger.info("Server cancelled")
except Exception as e:
logger.error(f"Server error: {e}")
traceback.print_exc()
raise
if __name__ == "__main__":
asyncio.run(main())
