from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import asyncio
import os
import logging
import sys
import json
from .tantivy_search_agent import TantivySearchAgent
# Configure logging
logging.basicConfig(
level=logging.ERROR,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stderr),
logging.FileHandler('jewish_library.log', encoding='utf-8')
]
)
logger = logging.getLogger('jewish_library')
# Initialize TantivySearchAgent with the index path
current_dir = os.path.dirname(os.path.abspath(__file__))
index_path = os.path.join(os.path.dirname(os.path.dirname(current_dir)), "index")
try:
search_agent = TantivySearchAgent(index_path)
logger.info("Search agent initialized")
except Exception as e:
logger.error(f"Failed to initialize search agent: {e}")
raise
server = Server("jewish_library")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
logger.debug("Handling list_tools request")
return [
types.Tool(
name="full_text_search",
description="Full text searching in the jewish library",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": """
Instructions for generating a query:
1. Boolean Operators:
- AND: term1 AND term2 (both required)
- OR: term1 OR term2 (either term)
- Multiple words default to OR operation (cloud network = cloud OR network)
- AND takes precedence over OR
- Example: Shabath AND (walk OR go)
2. Field-specific Terms:
- Field-specific terms: field:term
- Example: text:אדם AND reference:בראשית
- available fields: text, reference, topics
- text contains the text of the document
- reference contains the citation of the document, e.g. בראשית, פרק א
- topics contains the topics of the document. available topics includes: תנך, הלכה, מדרש, etc.
3. Required/Excluded Terms:
- Required (+): +term (must contain)
- Excluded (-): -term (must not contain)
- Example: +security cloud -deprecated
- Equivalent to: security AND cloud AND NOT deprecated
4. Phrase Search:
- Use quotes: "exact phrase"
- Both single/double quotes work
- Escape quotes with \\"
- Slop operator: "term1 term2"~N
- Example: "cloud security"~2
- the above will find "cloud framework and security "
- Prefix matching: "start of phrase"*
5. Wildcards:
- ? for single character
- * for any number of characters
- Example: sec?rity cloud*
6. Special Features:
- All docs: *
- Boost terms: term^2.0 (positive numbers only)
- Example: security^2.0 cloud
- the above will boost security by 2.0
Query Examples:
1. Basic: +שבת +חולה +אסור
2. Field-specific: text:סיני AND topics:תנך
3. Phrase with slop: "security framework"~2
4. Complex: +reference:בראשית +text:"הבל"^2.0 +(דמי OR דמים) -הבלים
6. Mixed: (text:"רבנו משה"^2.0 OR reference:"משנה תורה") AND topics:הלכה) AND text:"תורה המלך"~3 AND NOT topics:מדרש
Tips:
- Group complex expressions with parentheses
- Use quotes for exact phrases
- Add + for required terms, - for excluded terms
- Boost important terms with ^N
- use field-specific terms for better results.
- the corpus to search in is an ancient Hebrew corpus: Tora and Talmud. so Try to use ancient Hebrew terms and or Talmudic expressions and prevent modern words that are not common in talmudic texts
"""
},
"num_results": {
"type": "integer",
"description": "The maximum number of results to return (default: 25)",
"default": 25,
},
},
"required": ["query"],
},
),
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Tools can search the Jewish library and return formatted results.
"""
logger.debug(f"Handling call_tool request for {name} with arguments {arguments}")
try:
if not arguments:
raise ValueError("Missing arguments")
if name == "full_text_search":
try:
query = arguments.get("query")
if not query:
raise ValueError("Missing query parameter")
num_results = arguments.get("num_results", 25)
if not isinstance(num_results, int) or num_results <= 0:
raise ValueError("Invalid num_results parameter")
logger.info(f"Searching with query: {query}")
# Now do the actual search
logger.debug(f"Executing search with query: {query}")
results = await search_agent.search(query, num_results = num_results)
logger.debug(f"Search completed: {len(results)} results")
if not results or len(results) == 0:
logger.info("No results found")
return [types.TextContent(
type="text",
text="No results found"
)]
formatted_results = []
for result in results:
text = result.get('text', 'N/A')
reference = result.get('reference', 'N/A')
formatted_results.append(f"Reference: {reference}\nText: {text}\n")
logger.info(f"Found {len(formatted_results)} results")
response_text = "\n\n".join(formatted_results)
logger.debug(f"Response text: {response_text}")
return [
types.TextContent(
type="text",
text=response_text
)
]
except Exception as err:
logger.error(f"Search error: {err}", exc_info=True)
return [types.TextContent(
type="text",
text=f"Error: {str(err)}"
)]
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Tool execution error: {e}", exc_info=True)
return [types.TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def main():
try:
logger.info("Starting Jewish Library MCP server...")
# Run the server using stdin/stdout streams
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="jewish_library",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
raise
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Fatal error: {e}", exc_info=True)
raise