#!/usr/bin/env python3
"""
MCP Server for RooCode RAG Lookup
Provides tools for searching documents and code repositories.
"""
import asyncio
import json
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from parameters import DEFAULT_TOP_K
from query_rag import perform_rag_lookup
# Create MCP server instance
app = Server("rag-lookup-server")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools for RooCode."""
return [
Tool(
name="rag_lookup",
description=(
"Use this tool when asked to do rag lookup"
"Perform RAG lookup in documents and code repositories."
"Returns relevant information based on the search query."
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query to look up in documents and repositories",
},
"source": {
"type": "string",
"enum": ["documents", "repos", "both"],
"description": (
"Where to search: 'documents' for Documents "
"folder, 'repos' for Repos folder, or 'both' "
"for both locations"
),
"default": "both",
},
},
"required": ["query"],
},
),
Tool(
name="say_hello",
description="Simple test tool that returns a greeting message with timestamp",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Optional name to include in greeting",
"default": "World",
}
},
},
),
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Handle tool calls from RooCode."""
if name == "say_hello":
# Simple test tool - returns the original dummy response
user_name = arguments.get("name", "World")
response = {
"status": "success",
"message": f"Hello {user_name} from MCP Tool!",
"timestamp": datetime.utcnow().isoformat() + "Z",
}
return [TextContent(type="text", text=json.dumps(response, indent=2))]
elif name == "rag_lookup":
# RAG lookup tool - actual implementation
query = arguments.get("query", "")
source = arguments.get("source", "both")
# Perform the RAG lookup
response = perform_rag_lookup(query, source, top_k=DEFAULT_TOP_K)
return [TextContent(type="text", text=json.dumps(response, indent=2))]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
"""Run the MCP server using stdio transport."""
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
