MCP Sentiment Analysis Server

jimthompson5802: in folder mcp-swimrules, create file product_requirements_doc.md to build an agent that has a web UI and accesses tools using mcp to answer questions on swim rules. Ask questions that can help define the PRD document. GitHub Copilot: I'd like to help you create a comprehensive Product Requirements Document (PRD) for your swim rules agent. Before I create the document, let me ask some clarifying questions to ensure we build the right solution: ## Technical Architecture Questions: 1. **MCP Tools**: What specific swim rule tools do you envision? (e.g., rule lookup, violation checking, meet regulation validation, equipment compliance checking?) 2. **Data Sources**: What swim rule databases/sources should the agent access? (USA Swimming, FINA, NCAA, high school rules, etc.) 3. **Web Framework**: Do you have a preference for the web UI framework? (React, Vue, plain HTML/JS, or Python-based like Streamlit/FastAPI?) ## User Experience Questions: 4. **Target Users**: Who will primarily use this agent? (coaches, officials, swimmers, parents, meet directors?) 5. **Query Types**: What kinds of questions should users be able to ask? (rule interpretations, penalty assessments, equipment validation, meet procedures?) 6. **Response Format**: How should answers be presented? (text explanations, rule citations, visual aids, step-by-step guidance?) ## Functional Requirements: 7. **Real-time vs Batch**: Should the agent handle real-time rule questions during meets, or is it more for training/education? 8. **Rule Updates**: How should the system handle rule changes and updates? 9. **Complexity**: Should it handle simple rule lookups or complex scenario analysis involving multiple rules? ## Integration & Deployment: 10. **Authentication**: Do you need user authentication or can it be public? 11. **Deployment**: Where will this be hosted? (local, cloud, specific platform requirements?) 12. **Mobile**: Should it work well on mobile devices for poolside use? Based on your answers, I'll create a detailed PRD that aligns with your vision and technical requirements. Would you like to answer these questions, or should I proceed with reasonable assumptions and create a comprehensive PRD that you can then modify? jimthompson5802: 1 rule lookup, violation checking, meet regulation validation 2 USA Swimming 3. no preference 4. officials 5 all of the listed questions 6. rule citation and text explanation 7 training and education 8 Yes handle rule updates 9 both simple rule lookup and complex scenario analysis 10 no 11. Locally for now...should not preclue deployment to a different kind of environment 12. Mobile support not required. GitHub Copilot: Based on your answers, I'll create a comprehensive PRD for your swim rules agent. Let me create the document in the mcp-swimrules folder within your MCP testlab project. ## Solution Overview I'll create a Product Requirements Document that defines: 1. A web-based agent for swim rule queries and analysis 2. MCP integration for accessing USA Swimming rule databases 3. Support for officials with rule lookup, violation checking, and meet regulation validation 4. Both simple and complex rule analysis capabilities 5. Local deployment with cloud migration potential ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Creating a comprehensive PRD for the swim rules agent with MCP integration. ````markdown # Product Requirements Document: Swim Rules Agent ## 1. Executive Summary ### 1.1 Project Overview The Swim Rules Agent is a web-based application that provides officials with quick access to USA Swimming rules, violation checking, and meet regulation validation. The system leverages the Model Context Protocol (MCP) to access swim rule databases and provide intelligent rule analysis for training and educational purposes. ### 1.2 Business Objectives - Provide instant access to USA Swimming rules and regulations - Reduce rule interpretation errors during training sessions - Support officials' education and certification processes - Enable complex scenario analysis for rule application ### 1.3 Success Metrics - Query response time < 2 seconds for simple rule lookups - 95% accuracy in rule citation and interpretation - Support for 100+ concurrent users during training sessions - Zero downtime for rule updates ## 2. Product Scope ### 2.1 In Scope - Web-based user interface for rule queries - MCP integration for accessing USA Swimming rule databases - Rule lookup functionality with citation references - Swimming violation checking and validation - Meet regulation compliance verification - Complex scenario analysis involving multiple rules - Rule update management system - Local deployment with cloud migration readiness ### 2.2 Out of Scope - Mobile-specific applications - Real-time meet officiating integration - User authentication system (Phase 1) - Multi-organization rule support (Phase 1) - Automated rule enforcement during live meets ## 3. Target Users ### 3.1 Primary Users - **Swimming Officials**: Referees, starters, stroke & turn judges - **Official Trainers**: Instructors conducting certification programs - **Meet Directors**: Personnel managing swimming competitions ### 3.2 User Personas - **Experienced Official**: Needs quick rule clarification and complex scenario analysis - **Training Official**: Requires detailed explanations and educational context - **Meet Director**: Needs regulation validation and compliance checking ## 4. Functional Requirements ### 4.1 Core Features #### 4.1.1 Rule Lookup System - **Requirement**: Users can search for specific USA Swimming rules - **Acceptance Criteria**: - Full-text search across rule database - Search by rule number, keyword, or category - Return rule citation with complete text - Display related rules and cross-references - Search response time < 2 seconds #### 4.1.2 Violation Checking - **Requirement**: Analyze described swimming scenarios for rule violations - **Acceptance Criteria**: - Accept natural language scenario descriptions - Identify applicable rules and potential violations - Provide detailed explanation of violations - Suggest corrective actions or penalties - Handle multiple violation scenarios #### 4.1.3 Meet Regulation Validation - **Requirement**: Validate meet procedures against USA Swimming regulations - **Acceptance Criteria**: - Check meet format compliance - Validate timing and scoring procedures - Verify equipment and facility requirements - Provide compliance checklists - Generate regulation summary reports #### 4.1.4 Complex Scenario Analysis - **Requirement**: Analyze multi-faceted situations involving multiple rules - **Acceptance Criteria**: - Process scenarios with multiple rule interactions - Prioritize rule applications - Identify conflicting regulations - Provide step-by-step analysis - Support "what-if" scenario modeling ### 4.2 MCP Integration Features #### 4.2.1 Rule Database Access - **MCP Tools Required**: - `search_rules`: Full-text rule search functionality - `get_rule_by_number`: Retrieve specific rule by reference number - `get_related_rules`: Find related and cross-referenced rules - `validate_scenario`: Analyze scenarios for rule compliance #### 4.2.2 Data Management Tools - **MCP Tools Required**: - `update_rules`: Handle USA Swimming rule updates - `sync_database`: Synchronize with official rule sources - `validate_database`: Ensure data integrity - `get_rule_history`: Track rule changes and versions ### 4.3 User Interface Requirements #### 4.3.1 Query Interface - Clean, intuitive search interface - Support for natural language queries - Advanced search filters (category, rule type, date) - Query history and saved searches - Export functionality for results #### 4.3.2 Response Display - Clear rule citation format - Complete rule text with highlighting - Visual indicators for violation severity - Structured analysis for complex scenarios - Print-friendly formatting ### 4.4 Rule Update Management - **Requirement**: Automatically handle USA Swimming rule updates - **Acceptance Criteria**: - Detect new rule publications - Update database without service interruption - Maintain rule version history - Notify users of significant changes - Validate update integrity ## 5. Technical Requirements ### 5.1 Architecture - **Web Framework**: FastAPI or Flask (Python-based) - **MCP Integration**: FastMCP Python SDK - **Database**: PostgreSQL for rule storage - **Frontend**: React or Vue.js for interactive UI - **Transport**: HTTP-based MCP transport for web integration ### 5.2 Performance Requirements - Rule lookup response time: < 2 seconds - Complex analysis completion: < 10 seconds - Support 100 concurrent users - 99.9% uptime during training periods - Database update processing: < 5 minutes ### 5.3 Deployment Requirements - **Local Deployment**: Docker containerization - **Cloud Readiness**: Environment-agnostic configuration - **Scalability**: Horizontal scaling support - **Monitoring**: Application and performance monitoring - **Backup**: Automated database backup system ### 5.4 Data Requirements - USA Swimming Official Rules and Regulations - Technical Rules for competitive swimming - Administrative regulations - Historical rule versions and change logs - Rule interpretation guidelines ## 6. Non-Functional Requirements ### 6.1 Usability - Intuitive interface requiring minimal training - Consistent UI/UX patterns - Accessible design (WCAG 2.1 Level AA) - Responsive design for various screen sizes ### 6.2 Reliability - 99.9% system availability - Graceful error handling and recovery - Data consistency and integrity - Fault-tolerant MCP connections ### 6.3 Security - Secure data transmission (HTTPS) - Input validation and sanitization - Protection against common web vulnerabilities - Audit logging for system access ### 6.4 Maintainability - Modular, well-documented codebase - Automated testing suite - CI/CD pipeline support - Configuration management ## 7. Implementation Phases ### 7.1 Phase 1: Core Functionality (Weeks 1-4) - Basic rule lookup system - MCP server implementation - Simple web interface - Database setup and initial data load ### 7.2 Phase 2: Advanced Features (Weeks 5-8) - Violation checking functionality - Complex scenario analysis - Enhanced UI with advanced search - Rule update management system ### 7.3 Phase 3: Optimization (Weeks 9-10) - Performance optimization - Comprehensive testing - Documentation completion - Deployment preparation ## 8. Technical Specifications ### 8.1 MCP Server Architecture ```python # Example MCP tools structure @app.tool("search_rules") async def search_rules(query: str, category: str = None) -> SearchResults: """Search USA Swimming rules by query and optional category""" @app.tool("validate_scenario") async def validate_scenario(scenario: str) -> ViolationAnalysis: """Analyze swimming scenario for rule violations""" @app.tool("get_regulation_compliance") async def get_regulation_compliance(meet_type: str) -> ComplianceChecklist: """Get compliance requirements for specific meet types""" ``` ### 8.2 Database Schema - **rules**: Rule text, numbers, categories, effective dates - **rule_history**: Historical versions and changes - **violations**: Common violations and penalties - **scenarios**: Analyzed scenarios and outcomes ### 8.3 API Endpoints - `GET /api/search`: Rule search functionality - `POST /api/analyze`: Scenario analysis - `GET /api/rules/{id}`: Specific rule retrieval - `POST /api/validate`: Meet regulation validation ## 9. Risks and Mitigation ### 9.1 Technical Risks - **Risk**: MCP connection failures - **Mitigation**: Implement connection retry logic and fallback mechanisms - **Risk**: Database performance with large rule sets - **Mitigation**: Implement efficient indexing and caching strategies ### 9.2 Business Risks - **Risk**: USA Swimming rule format changes - **Mitigation**: Flexible parsing system and manual override capabilities - **Risk**: User adoption challenges - **Mitigation**: Comprehensive training materials and intuitive design ## 10. Success Criteria ### 10.1 Launch Criteria - All Phase 1 features implemented and tested - Database populated with current USA Swimming rules - Performance benchmarks met - User acceptance testing completed ### 10.2 Post-Launch Metrics - User engagement: 80% of officials use system monthly - Accuracy: 95% correct rule interpretations - Performance: 95% of queries resolve within SLA - Reliability: 99.9% uptime achievement ## 11. Future Enhancements ### 11.1 Potential Features - Multi-language support - Integration with other swimming organizations (FINA, NCAA) - Mobile application development - Real-time meet integration - AI-powered rule interpretation ### 11.2 Scalability Considerations - Microservices architecture migration - Cloud-native deployment options - API rate limiting and throttling - Advanced caching strategies --- **Document Version**: 1.0 **Last Updated**: [Current Date] **Owner**: Development Team **Stakeholders**: Swimming Officials, Training Organizations ```` This PRD provides a comprehensive foundation for building your swim rules agent with MCP integration. It addresses all your requirements while maintaining flexibility for future enhancements and deployment options. Made changes. jimthompson5802: Will RAG be part of this solution? GitHub Copilot: Looking at the current PRD, RAG (Retrieval-Augmented Generation) would be a valuable addition to enhance the swim rules agent's capabilities. Let me update the document to include RAG considerations. ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Adding RAG integration specifications to enhance rule interpretation and scenario analysis capabilities. ````markdown <!-- ...existing code... --> #### 4.1.4 Complex Scenario Analysis - **Requirement**: Analyze multi-faceted situations involving multiple rules using RAG-enhanced interpretation - **Acceptance Criteria**: - Process scenarios with multiple rule interactions using vector similarity search - Prioritize rule applications based on semantic relevance - Identify conflicting regulations through context-aware analysis - Provide step-by-step analysis with supporting rule citations - Support "what-if" scenario modeling with historical precedent lookup - Generate natural language explanations augmented by retrieved rule context ### 4.2 MCP Integration Features #### 4.2.1 Rule Database Access - **MCP Tools Required**: - `search_rules`: Full-text and semantic rule search functionality - `get_rule_by_number`: Retrieve specific rule by reference number - `get_related_rules`: Find related and cross-referenced rules using vector similarity - `validate_scenario`: Analyze scenarios for rule compliance with RAG-enhanced interpretation - `semantic_search`: Vector-based semantic search across rule embeddings - `get_rule_context`: Retrieve contextual information and historical interpretations #### 4.2.2 Data Management Tools - **MCP Tools Required**: - `update_rules`: Handle USA Swimming rule updates and re-generate embeddings - `sync_database`: Synchronize with official rule sources - `validate_database`: Ensure data integrity - `get_rule_history`: Track rule changes and versions - `update_embeddings`: Refresh vector embeddings for rule database - `index_scenarios`: Index analyzed scenarios for future retrieval #### 4.2.3 RAG Enhancement Tools - **MCP Tools Required**: - `generate_embeddings`: Create vector embeddings for rules and scenarios - `similarity_search`: Find semantically similar rules and precedents - `context_retrieval`: Retrieve relevant context for rule interpretation - `augment_response`: Enhance responses with retrieved contextual information ### 4.3 User Interface Requirements #### 4.3.1 Query Interface - Clean, intuitive search interface with semantic search capabilities - Support for natural language queries with intent recognition - Advanced search filters (category, rule type, date, similarity threshold) - Query history and saved searches with semantic clustering - Export functionality for results including retrieved context - Visual similarity indicators for related rules and scenarios #### 4.3.2 Response Display - Clear rule citation format with confidence scores - Complete rule text with semantic highlighting - Visual indicators for violation severity and rule relevance - Structured analysis for complex scenarios with supporting evidence - Print-friendly formatting with embedded context - Source attribution for retrieved contextual information ### 4.4 Rule Update Management - **Requirement**: Automatically handle USA Swimming rule updates with RAG pipeline refresh - **Acceptance Criteria**: - Detect new rule publications and trigger embedding updates - Update database without service interruption - Maintain rule version history with embedding versioning - Notify users of significant changes affecting semantic relationships - Validate update integrity including embedding quality - Re-index historical scenarios against updated rule base ## 5. Technical Requirements ### 5.1 Architecture - **Web Framework**: FastAPI for high-performance API endpoints - **MCP Integration**: FastMCP Python SDK with RAG-enhanced tools - **Database**: PostgreSQL for structured rule storage + Vector database (pgvector or Chroma) - **Frontend**: React or Vue.js for interactive UI with semantic search components - **Transport**: HTTP-based MCP transport for web integration - **RAG Components**: - Embedding model (e.g., sentence-transformers, OpenAI embeddings) - Vector store for rule and scenario embeddings - Retrieval system with semantic similarity ranking - Context augmentation pipeline ### 5.2 RAG-Specific Requirements - **Embedding Generation**: Sub-second embedding creation for new rules - **Vector Search**: < 500ms similarity search across 10,000+ rule embeddings - **Context Retrieval**: Retrieve top-k (3-5) most relevant rule contexts - **Response Augmentation**: Seamless integration of retrieved context into responses - **Embedding Quality**: Maintain cosine similarity > 0.8 for known related rules <!-- ...existing code... --> ## 8. Technical Specifications ### 8.1 MCP Server Architecture ```python # Example MCP tools structure with RAG integration @app.tool("search_rules") async def search_rules(query: str, category: str = None, use_semantic: bool = True) -> SearchResults: """Search USA Swimming rules by query with optional semantic search""" @app.tool("validate_scenario") async def validate_scenario(scenario: str, use_rag: bool = True) -> ViolationAnalysis: """Analyze swimming scenario for rule violations with RAG-enhanced interpretation""" @app.tool("get_regulation_compliance") async def get_regulation_compliance(meet_type: str) -> ComplianceChecklist: """Get compliance requirements for specific meet types""" @app.tool("semantic_search") async def semantic_search(query: str, top_k: int = 5) -> List[RuleMatch]: """Perform vector-based semantic search across rule embeddings""" @app.tool("get_rule_context") async def get_rule_context(rule_id: str) -> RuleContext: """Retrieve contextual information and historical interpretations for a rule""" ``` ### 8.2 Database Schema - **rules**: Rule text, numbers, categories, effective dates, embedding_vector - **rule_history**: Historical versions and changes with embedding evolution - **violations**: Common violations and penalties with semantic tags - **scenarios**: Analyzed scenarios and outcomes with embedding vectors - **rule_embeddings**: Vector representations of rules for semantic search - **scenario_embeddings**: Vector representations of scenarios for precedent lookup ### 8.3 RAG Pipeline Architecture - **Embedding Pipeline**: Rule text → sentence transformer → vector database - **Retrieval Pipeline**: Query → embedding → similarity search → context ranking - **Augmentation Pipeline**: Retrieved context + original query → enhanced response - **Feedback Loop**: User interactions → embedding refinement → improved retrieval <!-- ...existing code... --> ### 9.3 RAG-Specific Risks - **Risk**: Embedding model bias affecting rule interpretation - **Mitigation**: Regular embedding quality validation and model updates - **Risk**: Vector search returning irrelevant context - **Mitigation**: Implement similarity thresholds and relevance scoring - **Risk**: RAG responses lacking accuracy verification - **Mitigation**: Maintain traditional rule lookup as fallback and validation <!-- ...existing code... --> ## 11. Future Enhancements ### 11.1 Potential Features - Multi-language support with cross-lingual embeddings - Integration with other swimming organizations (FINA, NCAA) with federated RAG - Mobile application development with offline RAG capabilities - Real-time meet integration with context-aware rule assistance - AI-powered rule interpretation with advanced RAG techniques ### 11.2 RAG Enhancement Roadmap - **Phase 1**: Basic semantic search and context retrieval - **Phase 2**: Advanced RAG with reranking and query expansion - **Phase 3**: Multi-modal RAG including rule diagrams and videos - **Phase 4**: Conversational RAG with dialogue context maintenance <!-- ...existing code... --> ```` The RAG integration will significantly enhance the swim rules agent by: 1. **Semantic Understanding**: Enable natural language queries that understand intent beyond keyword matching 2. **Contextual Responses**: Provide richer answers by retrieving relevant historical interpretations and related rules 3. **Improved Accuracy**: Use vector similarity to find the most relevant rules even when exact terminology differs 4. **Precedent Lookup**: Retrieve similar scenarios and their outcomes to support decision-making 5. **Dynamic Learning**: Continuously improve responses based on usage patterns and feedback This RAG-enhanced approach will make the system much more intelligent and user-friendly for swimming officials. Made changes. jimthompson5802: For the user interface provide a diagram of the web page. The web page should consist of a input text box where the user can enter the scenario. An output text field that indicates decision on whether the scenario is "allowed" or a "disqualification". A text box explaining the rationale for the decision. And a text box that cites the specific rule identifiers justifying the decision and rationale. GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Adding a detailed web interface diagram to section 4.3 to specify the user interface layout and components. ````markdown <!-- ...existing code... --> ### 4.3 User Interface Requirements #### 4.3.1 Query Interface - Clean, intuitive search interface with semantic search capabilities - Support for natural language queries with intent recognition - Advanced search filters (category, rule type, date, similarity threshold) - Query history and saved searches with semantic clustering - Export functionality for results including retrieved context - Visual similarity indicators for related rules and scenarios #### 4.3.2 Web Interface Layout ``` ┌─────────────────────────────────────────────────────────────────────────────┐ │ Swim Rules Agent │ │ Official Scenario Analysis │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ Scenario Input: │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ Enter swimming scenario description... │ │ │ │ │ │ │ │ Example: "Swimmer did not touch the wall with both hands │ │ │ │ simultaneously during breaststroke turn" │ │ │ │ │ │ │ │ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │ │ [Analyze] │ │ [Clear Input] │ │ [Load Example] │ │ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ ├─────────────────────────────────────────────────────────────────────────────┤ │ ANALYSIS RESULTS │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ Decision: │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 🚫 DISQUALIFICATION │ │ │ │ Confidence: 95% │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ Rationale: │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ The described scenario constitutes a violation of breaststroke │ │ │ │ technique requirements. USA Swimming rules mandate that swimmers │ │ │ │ must touch the wall with both hands simultaneously during turns │ │ │ │ and finishes in breaststroke events. Failure to do so results in │ │ │ │ immediate disqualification as it provides an unfair competitive │ │ │ │ advantage and violates stroke technique standards. │ │ │ │ │ │ │ │ The violation is clear and unambiguous, requiring no official │ │ │ │ discretion in application. │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ Rule Citations: │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ Primary Rules: │ │ │ │ • 101.2.2 - Breaststroke Turn Requirements │ │ │ │ • 101.2.3 - Breaststroke Finish Requirements │ │ │ │ │ │ │ │ Supporting Rules: │ │ │ │ • 102.5.1 - Disqualification Procedures │ │ │ │ • 103.1.4 - Stroke Technique Violations │ │ │ │ │ │ │ │ Related Interpretations: │ │ │ │ • Official Interpretation 2023-BR-15: Simultaneous Touch Definition│ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │ │ [Export Report] │ │ [Save Scenario] │ │ [Print Results] │ │ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────┘ ``` #### 4.3.3 Interface Component Specifications **Scenario Input Area:** - Multi-line text area (minimum 4 rows, expandable) - Character limit: 2000 characters with live counter - Placeholder text with example scenarios - Auto-save draft functionality - Input validation for minimum meaningful content **Decision Display:** - Large, prominently displayed result banner - Color-coded status indicators: - 🟢 GREEN: "ALLOWED" / "NO VIOLATION" - 🔴 RED: "DISQUALIFICATION" / "VIOLATION" - 🟡 YELLOW: "CAUTION" / "DISCRETIONARY" - Confidence percentage display (RAG-generated confidence score) - Visual icons for quick recognition **Rationale Text Box:** - Read-only formatted text area - Supports rich text formatting for emphasis - Expandable/collapsible for long explanations - Word wrap enabled - Copy-to-clipboard functionality **Rule Citations Box:** - Structured list format with hierarchical display - Clickable rule numbers for detailed rule text - Categorized citations (Primary, Supporting, Related) - Link to official USA Swimming rule database - Export citations in standard format #### 4.3.4 Response Display - Clear rule citation format with confidence scores - Complete rule text with semantic highlighting - Visual indicators for violation severity and rule relevance - Structured analysis for complex scenarios with supporting evidence - Print-friendly formatting with embedded context - Source attribution for retrieved contextual information <!-- ...existing code... --> ```` This diagram provides a clear visual specification for the web interface that includes: 1. **Input Section**: Large text area for scenario description with helpful examples 2. **Decision Section**: Prominent, color-coded decision display with confidence scoring 3. **Rationale Section**: Detailed explanation area for the reasoning behind the decision 4. **Citations Section**: Structured display of relevant rule numbers and interpretations 5. **Action Buttons**: Export, save, and print functionality for practical use The layout is designed to be intuitive for swimming officials and supports their workflow during training and educational activities. Made changes. Similar code found with 5 license types jimthompson5802: Ensure the PRD support async operations GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Adding comprehensive async operation support throughout the technical architecture and specifications. ````markdown <!-- ...existing code... --> ### 4.4 Rule Update Management - **Requirement**: Automatically handle USA Swimming rule updates with RAG pipeline refresh using async operations - **Acceptance Criteria**: - Detect new rule publications and trigger embedding updates asynchronously - Update database without service interruption using background tasks - Maintain rule version history with embedding versioning via async workers - Notify users of significant changes affecting semantic relationships through async notifications - Validate update integrity including embedding quality using async validation pipelines - Re-index historical scenarios against updated rule base with async batch processing ## 5. Technical Requirements ### 5.1 Architecture - **Web Framework**: FastAPI for high-performance async API endpoints - **MCP Integration**: FastMCP Python SDK with async RAG-enhanced tools - **Database**: PostgreSQL with asyncpg for async operations + Vector database (pgvector or Chroma) with async drivers - **Frontend**: React or Vue.js for interactive UI with async semantic search components - **Transport**: HTTP-based MCP transport with async connection handling - **Async Components**: - AsyncIO event loop for concurrent request handling - Background task queues (Celery/Redis or FastAPI BackgroundTasks) - Async context managers for resource management - Connection pooling for database async operations - **RAG Components**: - Async embedding model interfaces (e.g., sentence-transformers, OpenAI embeddings) - Async vector store operations for rule and scenario embeddings - Async retrieval system with concurrent similarity ranking - Async context augmentation pipeline with streaming responses ### 5.2 Async-Specific Requirements - **Concurrent Request Handling**: Support 100+ simultaneous async requests - **Non-blocking Operations**: All database and MCP operations must be async/await compatible - **Background Task Processing**: Async queue processing for rule updates and embedding generation - **Async Resource Management**: Use AsyncExitStack for proper cleanup of async resources - **Streaming Responses**: Support async streaming for large rule analysis results - **Connection Pooling**: Async database connection pools with configurable limits ### 5.3 RAG-Specific Requirements - **Async Embedding Generation**: Non-blocking embedding creation for new rules - **Concurrent Vector Search**: Parallel similarity search across 10,000+ rule embeddings - **Async Context Retrieval**: Non-blocking retrieval of top-k (3-5) most relevant rule contexts - **Streaming Response Augmentation**: Async integration of retrieved context into streaming responses - **Async Embedding Quality Validation**: Background validation maintaining cosine similarity > 0.8 ### 5.4 Performance Requirements - Async rule lookup response time: < 2 seconds with concurrent handling - Async complex analysis completion: < 10 seconds with parallel processing - Support 100 concurrent async users without blocking - 99.9% uptime during training periods with async health checks - Async database update processing: < 5 minutes with background tasks ### 5.5 Deployment Requirements - **Local Deployment**: Docker containerization with async-capable Python runtime - **Cloud Readiness**: Environment-agnostic async configuration - **Async Scalability**: Horizontal scaling support with async load balancing - **Async Monitoring**: Application and performance monitoring with async metrics collection - **Async Backup**: Automated database backup system with non-blocking operations <!-- ...existing code... --> ## 8. Technical Specifications ### 8.1 MCP Server Architecture ```python # Example MCP tools structure with async RAG integration from fastmcp import FastMCP from typing import List import asyncio app = FastMCP("swim-rules-agent") @app.tool("search_rules") async def search_rules(query: str, category: str = None, use_semantic: bool = True) -> SearchResults: """Search USA Swimming rules by query with optional semantic search using async operations""" async with database_pool.acquire() as conn: # Async database operations pass @app.tool("validate_scenario") async def validate_scenario(scenario: str, use_rag: bool = True) -> ViolationAnalysis: """Analyze swimming scenario for rule violations with RAG-enhanced interpretation using async processing""" async with AsyncExitStack() as stack: # Async resource management embedding_service = await stack.enter_async_context(EmbeddingService()) vector_db = await stack.enter_async_context(VectorDatabase()) # Concurrent processing of multiple rule checks tasks = [ check_stroke_violations(scenario), check_timing_violations(scenario), check_equipment_violations(scenario) ] results = await asyncio.gather(*tasks) return aggregate_analysis(results) @app.tool("get_regulation_compliance") async def get_regulation_compliance(meet_type: str) -> ComplianceChecklist: """Get compliance requirements for specific meet types using async database queries""" async with database_pool.acquire() as conn: # Async compliance checking pass @app.tool("semantic_search") async def semantic_search(query: str, top_k: int = 5) -> List[RuleMatch]: """Perform vector-based semantic search across rule embeddings with async operations""" async with vector_store_client.session() as session: # Async vector similarity search pass @app.tool("get_rule_context") async def get_rule_context(rule_id: str) -> RuleContext: """Retrieve contextual information and historical interpretations for a rule using async operations""" async with database_pool.acquire() as conn: # Async context retrieval pass @app.tool("background_update_rules") async def background_update_rules() -> UpdateStatus: """Handle rule updates in background using async task processing""" # Schedule background task for rule updates task = asyncio.create_task(update_rules_async()) return {"status": "processing", "task_id": id(task)} # Async helper functions async def check_stroke_violations(scenario: str) -> List[Violation]: """Async stroke violation checking""" pass async def check_timing_violations(scenario: str) -> List[Violation]: """Async timing violation checking""" pass async def check_equipment_violations(scenario: str) -> List[Violation]: """Async equipment violation checking""" pass async def update_rules_async() -> None: """Background async rule update processing""" async with AsyncExitStack() as stack: db = await stack.enter_async_context(AsyncDatabase()) embedding_service = await stack.enter_async_context(AsyncEmbeddingService()) # Async rule update pipeline pass ``` ### 8.2 Async Database Schema Operations - **Async Connection Management**: Connection pooling with asyncpg for PostgreSQL - **Non-blocking Queries**: All database operations use async/await patterns - **Concurrent Transactions**: Support for concurrent rule updates and queries - **Async Indexing**: Background async indexing for vector embeddings - **Streaming Results**: Async generators for large result sets ### 8.3 Async RAG Pipeline Architecture ```python # Async RAG Pipeline Implementation class AsyncRAGPipeline: async def __aenter__(self): """Async context manager entry""" self.embedding_service = await AsyncEmbeddingService().connect() self.vector_store = await AsyncVectorStore().connect() self.database = await AsyncDatabase().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" await self.embedding_service.close() await self.vector_store.close() await self.database.close() async def process_query(self, query: str) -> EnhancedResponse: """Async RAG query processing with concurrent operations""" # Concurrent embedding generation and similarity search embedding_task = asyncio.create_task(self.embedding_service.embed(query)) embedding = await embedding_task # Concurrent context retrieval similarity_task = asyncio.create_task(self.vector_store.similarity_search(embedding)) rule_task = asyncio.create_task(self.database.get_related_rules(query)) similar_contexts, related_rules = await asyncio.gather(similarity_task, rule_task) # Async response augmentation return await self.augment_response(query, similar_contexts, related_rules) ``` ### 8.4 Async Client Architecture ```python # Example async client implementation class AsyncSwimRulesClient: def __init__(self): self.session: Optional[ClientSession] = None self.transport: Optional[Transport] = None self.exit_stack = AsyncExitStack() async def __aenter__(self): """Async context manager for resource management""" self.transport = await self.exit_stack.enter_async_context( get_http_transport("http://localhost:8000") ) self.session = await self.exit_stack.enter_async_context( ClientSession(self.transport) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Cleanup async resources""" await self.exit_stack.aclose() async def analyze_scenario(self, scenario: str) -> ViolationAnalysis: """Async scenario analysis with streaming response""" result = await self.session.call_tool("validate_scenario", {"scenario": scenario}) return ViolationAnalysis.from_dict(result.content[0].text) async def search_rules_concurrent(self, queries: List[str]) -> List[SearchResults]: """Concurrent rule searches""" tasks = [self.session.call_tool("search_rules", {"query": q}) for q in queries] results = await asyncio.gather(*tasks) return [SearchResults.from_dict(r.content[0].text) for r in results] ``` ### 8.5 Async API Endpoints ```python # FastAPI async endpoint examples from fastapi import FastAPI, BackgroundTasks from fastapi.responses import StreamingResponse app = FastAPI() @app.post("/api/analyze") async def analyze_scenario(scenario: ScenarioRequest, background_tasks: BackgroundTasks) -> ViolationResponse: """Async scenario analysis endpoint""" async with AsyncSwimRulesClient() as client: result = await client.analyze_scenario(scenario.text) # Schedule background task for logging/analytics background_tasks.add_task(log_analysis_async, scenario, result) return ViolationResponse.from_analysis(result) @app.get("/api/search/stream") async def stream_search_results(query: str) -> StreamingResponse: """Streaming search results with async processing""" async def generate_results(): async with AsyncSwimRulesClient() as client: async for result in client.stream_search_results(query): yield f"data: {result.json()}\n\n" return StreamingResponse(generate_results(), media_type="text/plain") @app.post("/api/rules/update") async def trigger_rule_update(background_tasks: BackgroundTasks) -> UpdateResponse: """Async rule update trigger""" task_id = await schedule_rule_update_async() background_tasks.add_task(monitor_update_progress, task_id) return UpdateResponse(task_id=task_id, status="processing") async def log_analysis_async(scenario: ScenarioRequest, result: ViolationAnalysis) -> None: """Background async logging""" async with AsyncDatabase() as db: await db.log_scenario_analysis(scenario, result) async def schedule_rule_update_async() -> str: """Schedule async rule update task""" task = asyncio.create_task(perform_rule_update()) return str(id(task)) ``` <!-- ...existing code... --> ### 9.4 Async-Specific Risks - **Risk**: Async resource leaks from improper cleanup - **Mitigation**: Use AsyncExitStack and proper async context managers throughout - **Risk**: Deadlocks in concurrent async operations - **Mitigation**: Implement timeout handling and avoid blocking operations in async contexts - **Risk**: Performance degradation under high concurrent load - **Mitigation**: Connection pooling, rate limiting, and async monitoring - **Risk**: Complex async debugging and error tracking - **Mitigation**: Comprehensive async logging and distributed tracing <!-- ...existing code... --> ### 11.2 Async Enhancement Roadmap - **Phase 1**: Basic async operations with FastMCP and FastAPI - **Phase 2**: Advanced async patterns with streaming and background tasks - **Phase 3**: Distributed async processing with message queues - **Phase 4**: Real-time async collaboration features with WebSocket support <!-- ...existing code... --> ```` The PRD now comprehensively supports async operations throughout the entire architecture, ensuring: 1. **Non-blocking Operations**: All database, MCP, and RAG operations use async/await patterns 2. **Concurrent Processing**: Multiple requests and background tasks can run simultaneously 3. **Resource Management**: Proper async context managers prevent resource leaks 4. **Scalability**: Async architecture supports high concurrent user loads 5. **Performance**: Background tasks and streaming responses improve user experience 6. **Error Handling**: Async-aware error handling and cleanup mechanisms Made changes. jimthompson5802: do not update the PRD. Is it possible to use a sqlite3 database instead of postgres for this application? GitHub Copilot: Yes, it's definitely possible to use SQLite3 instead of PostgreSQL for this swim rules application. Here are the key considerations: ## SQLite3 Advantages for This Use Case: 1. **Simpler Deployment**: No separate database server needed - single file database 2. **Local Development**: Perfect for the "local deployment" requirement in your PRD 3. **Zero Configuration**: No database setup or administration required 4. **Async Support**: `aiosqlite` provides full async/await compatibility 5. **Vector Extensions**: SQLite supports vector similarity search via extensions like `sqlite-vss` or `sqlite-vec` ## Technical Feasibility: **Async Operations**: ✅ Fully supported - `aiosqlite` provides async database operations - Works seamlessly with FastAPI and AsyncExitStack patterns **Vector Search/RAG**: ✅ Supported with extensions - `sqlite-vss` for vector similarity search - `sqlite-vec` for lightweight vector operations - Can handle 10,000+ embeddings efficiently **Performance**: ✅ Adequate for requirements - SQLite can handle 100+ concurrent readers - Sub-second query performance for rule lookups - Suitable for the specified performance requirements **Scalability**: ⚠️ Some limitations - Single writer at a time (but rarely an issue for read-heavy workloads) - File-based, so horizontal scaling requires application-level sharding - Still suitable for the 100 concurrent user requirement ## Recommended Architecture Changes: ```python # Instead of PostgreSQL + pgvector: DATABASE_URL = "sqlite+aiosqlite:///./swim_rules.db" # Vector search with sqlite-vss: async with aiosqlite.connect("swim_rules.db") as db: await db.enable_load_extension(True) await db.load_extension("vss0") # Vector similarity search operations ``` ## Migration Path: The PRD's "cloud readiness" requirement is still achievable: - SQLite for local/development - Easy migration to PostgreSQL for cloud deployment - Same async patterns work with both databases SQLite3 would be an excellent choice for this application, especially given the local deployment focus and the relatively modest scale requirements. jimthompson5802: update prd to use sqlite3 instead of postgres GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Updating the PRD to use SQLite3 instead of PostgreSQL for better local deployment and simplified architecture. ````markdown <!-- ...existing code... --> ## 5. Technical Requirements ### 5.1 Architecture - **Web Framework**: FastAPI for high-performance async API endpoints - **MCP Integration**: FastMCP Python SDK with async RAG-enhanced tools - **Database**: SQLite3 with aiosqlite for async operations + sqlite-vss extension for vector similarity search - **Frontend**: React or Vue.js for interactive UI with async semantic search components - **Transport**: HTTP-based MCP transport with async connection handling - **Async Components**: - AsyncIO event loop for concurrent request handling - Background task queues (Celery/Redis or FastAPI BackgroundTasks) - Async context managers for resource management - Connection pooling for database async operations - **RAG Components**: - Async embedding model interfaces (e.g., sentence-transformers, OpenAI embeddings) - SQLite-vss for vector embeddings and similarity search - Async retrieval system with concurrent similarity ranking - Async context augmentation pipeline with streaming responses ### 5.2 Async-Specific Requirements - **Concurrent Request Handling**: Support 100+ simultaneous async requests with SQLite connection pooling - **Non-blocking Operations**: All database and MCP operations must be async/await compatible using aiosqlite - **Background Task Processing**: Async queue processing for rule updates and embedding generation - **Async Resource Management**: Use AsyncExitStack for proper cleanup of async resources - **Streaming Responses**: Support async streaming for large rule analysis results - **Connection Pooling**: Async SQLite connection pools with configurable limits ### 5.3 RAG-Specific Requirements - **Async Embedding Generation**: Non-blocking embedding creation for new rules using SQLite transactions - **Concurrent Vector Search**: Parallel similarity search across 10,000+ rule embeddings using sqlite-vss - **Async Context Retrieval**: Non-blocking retrieval of top-k (3-5) most relevant rule contexts - **Streaming Response Augmentation**: Async integration of retrieved context into streaming responses - **Async Embedding Quality Validation**: Background validation maintaining cosine similarity > 0.8 ### 5.4 Performance Requirements - Async rule lookup response time: < 2 seconds with concurrent handling - Async complex analysis completion: < 10 seconds with parallel processing - Support 100 concurrent async users without blocking (SQLite handles multiple readers efficiently) - 99.9% uptime during training periods with async health checks - Async database update processing: < 5 minutes with background tasks ### 5.5 Deployment Requirements - **Local Deployment**: Docker containerization with SQLite database file mounting - **Cloud Readiness**: Environment-agnostic async configuration with SQLite file backup to cloud storage - **Async Scalability**: Horizontal scaling support with SQLite file replication - **Async Monitoring**: Application and performance monitoring with async metrics collection - **Async Backup**: Automated SQLite database backup system with non-blocking file operations ### 5.6 Data Requirements - USA Swimming Official Rules and Regulations stored in SQLite tables - Technical Rules for competitive swimming with full-text search indexes - Administrative regulations with vector embeddings in sqlite-vss tables - Historical rule versions and change logs in versioned SQLite tables - Rule interpretation guidelines with semantic search capabilities <!-- ...existing code... --> ## 8. Technical Specifications ### 8.1 MCP Server Architecture ```python # Example MCP tools structure with async SQLite and RAG integration from fastmcp import FastMCP from typing import List import asyncio import aiosqlite import sqlite3 app = FastMCP("swim-rules-agent") # SQLite database configuration DATABASE_PATH = "swim_rules.db" @app.tool("search_rules") async def search_rules(query: str, category: str = None, use_semantic: bool = True) -> SearchResults: """Search USA Swimming rules by query with optional semantic search using async SQLite operations""" async with aiosqlite.connect(DATABASE_PATH) as db: # Enable sqlite-vss extension for vector search await db.enable_load_extension(True) await db.load_extension("vss0") if use_semantic: # Vector similarity search using sqlite-vss cursor = await db.execute(""" SELECT rule_id, rule_text, distance FROM vss_rules WHERE vss_search(embedding, vss_search_params(?, 5)) ORDER BY distance ASC """, (query,)) else: # Traditional full-text search cursor = await db.execute(""" SELECT rule_id, rule_text FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank """, (query,)) results = await cursor.fetchall() return SearchResults(results=results) @app.tool("validate_scenario") async def validate_scenario(scenario: str, use_rag: bool = True) -> ViolationAnalysis: """Analyze swimming scenario for rule violations with RAG-enhanced interpretation using async SQLite processing""" async with AsyncExitStack() as stack: # Async resource management with SQLite connections db = await stack.enter_async_context(aiosqlite.connect(DATABASE_PATH)) await db.enable_load_extension(True) await db.load_extension("vss0") # Concurrent processing of multiple rule checks tasks = [ check_stroke_violations(db, scenario), check_timing_violations(db, scenario), check_equipment_violations(db, scenario) ] results = await asyncio.gather(*tasks) return aggregate_analysis(results) @app.tool("get_regulation_compliance") async def get_regulation_compliance(meet_type: str) -> ComplianceChecklist: """Get compliance requirements for specific meet types using async SQLite queries""" async with aiosqlite.connect(DATABASE_PATH) as db: cursor = await db.execute(""" SELECT requirement_id, requirement_text, category FROM compliance_requirements WHERE meet_type = ? OR meet_type = 'all' ORDER BY priority DESC """, (meet_type,)) requirements = await cursor.fetchall() return ComplianceChecklist(requirements=requirements) @app.tool("semantic_search") async def semantic_search(query: str, top_k: int = 5) -> List[RuleMatch]: """Perform vector-based semantic search across rule embeddings with async SQLite operations""" async with aiosqlite.connect(DATABASE_PATH) as db: await db.enable_load_extension(True) await db.load_extension("vss0") cursor = await db.execute(""" SELECT rule_id, rule_text, rule_number, distance FROM vss_rules WHERE vss_search(embedding, vss_search_params(?, ?)) ORDER BY distance ASC """, (query, top_k)) matches = await cursor.fetchall() return [RuleMatch(rule_id=m[0], text=m[1], number=m[2], similarity=1-m[3]) for m in matches] @app.tool("get_rule_context") async def get_rule_context(rule_id: str) -> RuleContext: """Retrieve contextual information and historical interpretations for a rule using async SQLite operations""" async with aiosqlite.connect(DATABASE_PATH) as db: # Get rule details and related context cursor = await db.execute(""" SELECT r.rule_text, r.rule_number, r.category, GROUP_CONCAT(ri.interpretation_text) as interpretations, GROUP_CONCAT(rr.related_rule_id) as related_rules FROM rules r LEFT JOIN rule_interpretations ri ON r.rule_id = ri.rule_id LEFT JOIN rule_relationships rr ON r.rule_id = rr.rule_id WHERE r.rule_id = ? GROUP BY r.rule_id """, (rule_id,)) context = await cursor.fetchone() return RuleContext.from_tuple(context) @app.tool("background_update_rules") async def background_update_rules() -> UpdateStatus: """Handle rule updates in background using async SQLite task processing""" # Schedule background task for rule updates task = asyncio.create_task(update_rules_async()) return {"status": "processing", "task_id": id(task)} # Async helper functions for SQLite operations async def check_stroke_violations(db: aiosqlite.Connection, scenario: str) -> List[Violation]: """Async stroke violation checking using SQLite""" cursor = await db.execute(""" SELECT violation_id, violation_text, severity FROM stroke_violations sv JOIN rules r ON sv.rule_id = r.rule_id WHERE ? LIKE '%' || sv.trigger_pattern || '%' """, (scenario,)) violations = await cursor.fetchall() return [Violation(id=v[0], text=v[1], severity=v[2]) for v in violations] async def check_timing_violations(db: aiosqlite.Connection, scenario: str) -> List[Violation]: """Async timing violation checking using SQLite""" cursor = await db.execute(""" SELECT violation_id, violation_text, severity FROM timing_violations tv JOIN rules r ON tv.rule_id = r.rule_id WHERE ? LIKE '%' || tv.trigger_pattern || '%' """, (scenario,)) violations = await cursor.fetchall() return [Violation(id=v[0], text=v[1], severity=v[2]) for v in violations] async def check_equipment_violations(db: aiosqlite.Connection, scenario: str) -> List[Violation]: """Async equipment violation checking using SQLite""" cursor = await db.execute(""" SELECT violation_id, violation_text, severity FROM equipment_violations ev JOIN rules r ON ev.rule_id = r.rule_id WHERE ? LIKE '%' || ev.trigger_pattern || '%' """, (scenario,)) violations = await cursor.fetchall() return [Violation(id=v[0], text=v[1], severity=v[2]) for v in violations] async def update_rules_async() -> None: """Background async rule update processing with SQLite""" async with aiosqlite.connect(DATABASE_PATH) as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Update rules and regenerate embeddings await db.execute("DELETE FROM rules WHERE version < ?", (new_version,)) await db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?)", new_rules) # Update vector embeddings await db.enable_load_extension(True) await db.load_extension("vss0") await db.execute("DELETE FROM vss_rules") await db.executemany("INSERT INTO vss_rules VALUES (?, ?)", rule_embeddings) await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") raise e ``` ### 8.2 Async SQLite Database Schema Operations - **Async Connection Management**: Connection pooling with aiosqlite for SQLite operations - **Non-blocking Queries**: All database operations use async/await patterns with aiosqlite - **Concurrent Transactions**: SQLite WAL mode for concurrent readers with single writer - **Async Vector Indexing**: Background async indexing for vector embeddings using sqlite-vss - **Streaming Results**: Async generators for large result sets from SQLite queries ### 8.3 Async RAG Pipeline Architecture ```python # Async RAG Pipeline Implementation with SQLite class AsyncSQLiteRAGPipeline: def __init__(self, db_path: str = DATABASE_PATH): self.db_path = db_path self.embedding_service = None self.db_connection = None async def __aenter__(self): """Async context manager entry with SQLite connection""" self.db_connection = await aiosqlite.connect(self.db_path) await self.db_connection.enable_load_extension(True) await self.db_connection.load_extension("vss0") # Configure SQLite for optimal performance await self.db_connection.execute("PRAGMA journal_mode=WAL") await self.db_connection.execute("PRAGMA synchronous=NORMAL") await self.db_connection.execute("PRAGMA cache_size=10000") self.embedding_service = await AsyncEmbeddingService().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" if self.embedding_service: await self.embedding_service.close() if self.db_connection: await self.db_connection.close() async def process_query(self, query: str) -> EnhancedResponse: """Async RAG query processing with concurrent SQLite operations""" # Concurrent embedding generation and similarity search embedding_task = asyncio.create_task(self.embedding_service.embed(query)) embedding = await embedding_task # Concurrent context retrieval using SQLite similarity_task = asyncio.create_task(self._similarity_search(embedding)) rule_task = asyncio.create_task(self._get_related_rules(query)) similar_contexts, related_rules = await asyncio.gather(similarity_task, rule_task) # Async response augmentation return await self.augment_response(query, similar_contexts, related_rules) async def _similarity_search(self, embedding: List[float]) -> List[RuleContext]: """Perform vector similarity search using sqlite-vss""" cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, distance FROM vss_rules WHERE vss_search(embedding, vss_search_params(?, 5)) ORDER BY distance ASC """, (embedding,)) results = await cursor.fetchall() return [RuleContext(id=r[0], text=r[1], similarity=1-r[2]) for r in results] async def _get_related_rules(self, query: str) -> List[Rule]: """Get related rules using full-text search""" cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, rule_number FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank LIMIT 10 """, (query,)) results = await cursor.fetchall() return [Rule(id=r[0], text=r[1], number=r[2]) for r in results] ``` ### 8.4 Async Client Architecture ```python # Example async client implementation with SQLite awareness class AsyncSwimRulesClient: def __init__(self, db_path: str = DATABASE_PATH): self.db_path = db_path self.session: Optional[ClientSession] = None self.transport: Optional[Transport] = None self.exit_stack = AsyncExitStack() async def __aenter__(self): """Async context manager for resource management""" self.transport = await self.exit_stack.enter_async_context( get_http_transport("http://localhost:8000") ) self.session = await self.exit_stack.enter_async_context( ClientSession(self.transport) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Cleanup async resources""" await self.exit_stack.aclose() async def analyze_scenario(self, scenario: str) -> ViolationAnalysis: """Async scenario analysis with streaming response""" result = await self.session.call_tool("validate_scenario", {"scenario": scenario}) return ViolationAnalysis.from_dict(result.content[0].text) async def search_rules_concurrent(self, queries: List[str]) -> List[SearchResults]: """Concurrent rule searches leveraging SQLite's concurrent read capability""" tasks = [self.session.call_tool("search_rules", {"query": q}) for q in queries] results = await asyncio.gather(*tasks) return [SearchResults.from_dict(r.content[0].text) for r in results] async def backup_database(self, backup_path: str) -> bool: """Create async backup of SQLite database""" async with aiosqlite.connect(self.db_path) as source: async with aiosqlite.connect(backup_path) as backup: await source.backup(backup) return True ``` ### 8.5 SQLite Database Schema ```sql -- Core rules table with full-text search CREATE TABLE rules ( rule_id TEXT PRIMARY KEY, rule_number TEXT NOT NULL, rule_text TEXT NOT NULL, category TEXT NOT NULL, effective_date DATE, version INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Full-text search virtual table CREATE VIRTUAL TABLE rules_fts USING fts5( rule_id UNINDEXED, rule_number, rule_text, category, content='rules', content_rowid='rowid' ); -- Vector embeddings table using sqlite-vss CREATE VIRTUAL TABLE vss_rules USING vss0( embedding(1536), -- Dimension for OpenAI embeddings rule_id TEXT PRIMARY KEY ); -- Rule relationships and cross-references CREATE TABLE rule_relationships ( relationship_id INTEGER PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), related_rule_id TEXT REFERENCES rules(rule_id), relationship_type TEXT NOT NULL, strength REAL DEFAULT 1.0 ); -- Historical rule versions CREATE TABLE rule_history ( history_id INTEGER PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), previous_text TEXT, change_description TEXT, changed_date DATE, version INTEGER ); -- Violation patterns and triggers CREATE TABLE violations ( violation_id TEXT PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), violation_type TEXT NOT NULL, severity TEXT NOT NULL, penalty_description TEXT, trigger_patterns TEXT -- JSON array of pattern strings ); -- Analyzed scenarios for learning CREATE TABLE scenarios ( scenario_id TEXT PRIMARY KEY, scenario_text TEXT NOT NULL, analysis_result TEXT, -- JSON result applicable_rules TEXT, -- JSON array of rule_ids created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, embedding BLOB -- Serialized embedding vector ); -- Performance indexes CREATE INDEX idx_rules_category ON rules(category); CREATE INDEX idx_rules_version ON rules(version); CREATE INDEX idx_violations_type ON violations(violation_type); CREATE INDEX idx_scenarios_created ON scenarios(created_at); ``` <!-- ...existing code... --> ```` The PRD has been updated to use SQLite3 instead of PostgreSQL, providing: 1. **Simplified Architecture**: Single-file database with no external database server requirements 2. **Async Support**: Full aiosqlite integration for non-blocking operations 3. **Vector Search**: sqlite-vss extension for RAG functionality 4. **Better Local Deployment**: No database setup required, perfect for the local deployment requirement 5. **Cloud Migration Ready**: Easy to migrate to PostgreSQL later if needed 6. **Performance**: Suitable for the specified 100 concurrent users and response time requirements Made changes. jimthompson5802: do not update the PRD. What is the process to populate the database with usa swimming rules? GitHub Copilot: The process to populate the SQLite database with USA Swimming rules would involve several steps: ## Data Acquisition 1. **Official USA Swimming Rule Sources**: - USA Swimming Official Rules & Regulations PDF (published annually) - Technical Rules for competitive swimming - Administrative regulations and interpretations - Rule change bulletins and updates 2. **Data Access Methods**: - **Manual Download**: Download PDFs from USA Swimming website - **Web Scraping**: Automated extraction from official website (if permitted) - **API Access**: If USA Swimming provides rule APIs (unlikely) - **Partnership**: Direct data sharing agreement with USA Swimming ## Data Processing Pipeline 1. **PDF Text Extraction**: ```python # Using PyPDF2 or pdfplumber import pdfplumber async def extract_rules_from_pdf(pdf_path: str) -> List[RuleData]: """Extract structured rule data from USA Swimming PDF""" rules = [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text = page.extract_text() # Parse rule numbers, titles, and content parsed_rules = parse_rule_text(text) rules.extend(parsed_rules) return rules ``` 2. **Rule Structure Parsing**: - Parse rule numbers (e.g., "101.2.2", "102.5.1") - Extract rule titles and categories - Identify rule hierarchies and relationships - Parse effective dates and versions 3. **Data Cleaning & Normalization**: - Remove formatting artifacts - Standardize rule numbering - Identify cross-references between rules - Clean inconsistent text formatting ## Database Population Process 1. **Schema Creation**: ```python async def create_database_schema(): """Create SQLite database with tables and indexes""" async with aiosqlite.connect(DATABASE_PATH) as db: # Create all tables from the schema await db.execute(CREATE_RULES_TABLE) await db.execute(CREATE_FTS_TABLE) await db.execute(CREATE_VSS_TABLE) # ... other tables ``` 2. **Rule Insertion**: ```python async def populate_rules_table(rules_data: List[RuleData]): """Insert parsed rule data into database""" async with aiosqlite.connect(DATABASE_PATH) as db: await db.executemany(""" INSERT INTO rules (rule_id, rule_number, rule_text, category, effective_date) VALUES (?, ?, ?, ?, ?) """, rules_data) ``` 3. **Embedding Generation**: ```python async def generate_rule_embeddings(): """Generate vector embeddings for semantic search""" async with aiosqlite.connect(DATABASE_PATH) as db: # Get all rules cursor = await db.execute("SELECT rule_id, rule_text FROM rules") rules = await cursor.fetchall() # Generate embeddings for rule_id, rule_text in rules: embedding = await embedding_service.embed(rule_text) await db.execute(""" INSERT INTO vss_rules (rule_id, embedding) VALUES (?, ?) """, (rule_id, embedding)) ``` ## Implementation Strategy 1. **Initial Setup Script**: ```bash # One-time database initialization python scripts/init_database.py --pdf-path usa_swimming_rules_2024.pdf ``` 2. **Automated Updates**: ```python # Scheduled rule updates (quarterly/annually) async def check_for_rule_updates(): """Check USA Swimming website for new rule publications""" # Compare current version with downloaded version # Trigger update process if new version found ``` 3. **Data Validation**: - Verify rule completeness - Check for parsing errors - Validate cross-references - Test embedding quality ## Challenges & Solutions 1. **PDF Parsing Complexity**: - **Challenge**: Complex formatting, tables, diagrams - **Solution**: Multiple parsing strategies, manual review process 2. **Rule Relationships**: - **Challenge**: Implicit cross-references in text - **Solution**: NLP-based relationship extraction, manual curation 3. **Version Management**: - **Challenge**: Tracking rule changes over time - **Solution**: Diff analysis between versions, change tracking 4. **Legal Compliance**: - **Challenge**: Copyright and usage rights - **Solution**: Ensure compliance with USA Swimming terms, consider partnership ## Recommended Approach 1. **Phase 1**: Manual extraction and database setup with sample rules 2. **Phase 2**: Automated PDF processing pipeline 3. **Phase 3**: Integration with USA Swimming data sources 4. **Phase 4**: Automated update and monitoring system The key is starting with a robust parsing and validation system that can handle the specific formatting of USA Swimming rule documents. jimthompson5802: assume the rules will sourced from a pdf document. update the PRD with the process to ingest the rules from a pdf document. GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Adding comprehensive PDF rule ingestion process to the PRD technical specifications. ````markdown <!-- ...existing code... --> ### 5.6 Data Requirements - USA Swimming Official Rules and Regulations sourced from PDF documents - Technical Rules for competitive swimming with full-text search indexes - Administrative regulations with vector embeddings in sqlite-vss tables - Historical rule versions and change logs in versioned SQLite tables - Rule interpretation guidelines with semantic search capabilities ### 5.7 PDF Rule Ingestion Requirements - **PDF Processing**: Automated extraction of structured rule data from USA Swimming PDF documents - **Rule Parsing**: Intelligent parsing of rule numbers, titles, text, and hierarchical relationships - **Data Validation**: Comprehensive validation of extracted rule data for completeness and accuracy - **Version Management**: Track rule document versions and handle incremental updates - **Error Handling**: Robust error handling for PDF parsing failures and data inconsistencies - **Async Processing**: Non-blocking PDF processing with progress tracking and status updates <!-- ...existing code... --> ## 8. Technical Specifications ### 8.1 PDF Rule Ingestion Pipeline ```python # PDF Rule Ingestion Pipeline with async processing from fastmcp import FastMCP from typing import List, Dict, Optional, Tuple import asyncio import aiosqlite import pdfplumber import re import json from dataclasses import dataclass from pathlib import Path from contextlib import AsyncExitStack @dataclass class RuleData: """Data class for parsed rule information""" rule_id: str rule_number: str rule_title: str rule_text: str category: str section: str subsection: Optional[str] = None effective_date: Optional[str] = None version: int = 1 @dataclass class IngestionStatus: """Status tracking for PDF ingestion process""" total_pages: int processed_pages: int total_rules: int processed_rules: int errors: List[str] warnings: List[str] start_time: str status: str # 'processing', 'completed', 'failed' class PdfRuleIngestionPipeline: """Async PDF rule ingestion pipeline""" def __init__(self, db_path: str = DATABASE_PATH): self.db_path = db_path self.embedding_service = None self.rule_patterns = { 'rule_number': re.compile(r'^(\d+(?:\.\d+)*)\s+([A-Z][^.]*?)(?:\s*—\s*(.*))?$'), 'section_header': re.compile(r'^ARTICLE\s+(\d+)\s*[—-]\s*(.+)$'), 'subsection': re.compile(r'^(\d+\.\d+)\s+(.+)$'), 'effective_date': re.compile(r'Effective\s+Date:\s*(.+)', re.IGNORECASE) } async def __aenter__(self): """Async context manager entry""" self.embedding_service = await AsyncEmbeddingService().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" if self.embedding_service: await self.embedding_service.close() async def ingest_pdf(self, pdf_path: Path, version: int = 1) -> IngestionStatus: """ Main ingestion method for processing USA Swimming rules PDF Args: pdf_path: Path to the USA Swimming rules PDF document version: Version number for this rule set Returns: IngestionStatus: Status object with processing results """ status = IngestionStatus( total_pages=0, processed_pages=0, total_rules=0, processed_rules=0, errors=[], warnings=[], start_time=asyncio.get_event_loop().time(), status='processing' ) try: # Extract text from PDF extracted_data = await self._extract_pdf_content(pdf_path, status) # Parse rules from extracted text parsed_rules = await self._parse_rules(extracted_data, status) # Validate parsed rules validated_rules = await self._validate_rules(parsed_rules, status) # Store rules in database await self._store_rules(validated_rules, version, status) # Generate embeddings await self._generate_embeddings(validated_rules, status) status.status = 'completed' except Exception as e: status.errors.append(f"Ingestion failed: {str(e)}") status.status = 'failed' raise return status async def _extract_pdf_content(self, pdf_path: Path, status: IngestionStatus) -> List[Dict]: """Extract structured content from PDF pages""" extracted_pages = [] with pdfplumber.open(pdf_path) as pdf: status.total_pages = len(pdf.pages) for page_num, page in enumerate(pdf.pages): try: # Extract text with layout preservation text = page.extract_text(layout=True) # Extract tables if present tables = page.extract_tables() # Extract metadata page_data = { 'page_number': page_num + 1, 'text': text, 'tables': tables, 'bbox': page.bbox, 'width': page.width, 'height': page.height } extracted_pages.append(page_data) status.processed_pages += 1 # Yield control for async processing if page_num % 10 == 0: await asyncio.sleep(0) except Exception as e: status.warnings.append(f"Error processing page {page_num + 1}: {str(e)}") return extracted_pages async def _parse_rules(self, extracted_pages: List[Dict], status: IngestionStatus) -> List[RuleData]: """Parse rules from extracted PDF content""" rules = [] current_section = "" current_category = "" effective_date = None for page_data in extracted_pages: text_lines = page_data['text'].split('\n') for line_num, line in enumerate(text_lines): line = line.strip() if not line: continue try: # Check for section headers section_match = self.rule_patterns['section_header'].match(line) if section_match: current_section = section_match.group(1) current_category = section_match.group(2).strip() continue # Check for effective date date_match = self.rule_patterns['effective_date'].search(line) if date_match: effective_date = date_match.group(1).strip() continue # Check for rule number and content rule_match = self.rule_patterns['rule_number'].match(line) if rule_match: rule_number = rule_match.group(1) rule_title = rule_match.group(2).strip() # Extract rule text (may span multiple lines) rule_text = await self._extract_rule_text( text_lines, line_num, page_data['page_number'] ) rule_data = RuleData( rule_id=f"rule_{rule_number.replace('.', '_')}", rule_number=rule_number, rule_title=rule_title, rule_text=rule_text, category=current_category, section=current_section, effective_date=effective_date, version=1 ) rules.append(rule_data) status.total_rules += 1 except Exception as e: status.warnings.append( f"Error parsing line {line_num} on page {page_data['page_number']}: {str(e)}" ) # Yield control for async processing await asyncio.sleep(0) return rules async def _extract_rule_text(self, text_lines: List[str], start_line: int, page_number: int) -> str: """Extract complete rule text that may span multiple lines""" rule_text_lines = [] # Start from the line after the rule number for i in range(start_line + 1, len(text_lines)): line = text_lines[i].strip() # Stop if we hit another rule number or section header if (self.rule_patterns['rule_number'].match(line) or self.rule_patterns['section_header'].match(line) or line.startswith('ARTICLE')): break # Skip empty lines at the beginning if not rule_text_lines and not line: continue rule_text_lines.append(line) return ' '.join(rule_text_lines).strip() async def _validate_rules(self, rules: List[RuleData], status: IngestionStatus) -> List[RuleData]: """Validate parsed rules for completeness and accuracy""" validated_rules = [] for rule in rules: try: # Validate rule number format if not re.match(r'^\d+(?:\.\d+)*$', rule.rule_number): status.warnings.append(f"Invalid rule number format: {rule.rule_number}") continue # Validate required fields if not rule.rule_text or len(rule.rule_text.strip()) < 10: status.warnings.append(f"Rule {rule.rule_number} has insufficient text content") continue # Validate rule title if not rule.rule_title: status.warnings.append(f"Rule {rule.rule_number} missing title") continue validated_rules.append(rule) status.processed_rules += 1 except Exception as e: status.errors.append(f"Validation error for rule {rule.rule_number}: {str(e)}") return validated_rules async def _store_rules(self, rules: List[RuleData], version: int, status: IngestionStatus): """Store validated rules in SQLite database""" async with aiosqlite.connect(self.db_path) as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Clear existing rules for this version await db.execute("DELETE FROM rules WHERE version = ?", (version,)) # Insert new rules rule_tuples = [ ( rule.rule_id, rule.rule_number, rule.rule_text, rule.category, rule.effective_date, version, rule.rule_title, rule.section ) for rule in rules ] await db.executemany(""" INSERT INTO rules (rule_id, rule_number, rule_text, category, effective_date, version, rule_title, section) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, rule_tuples) # Update full-text search index await db.execute("INSERT INTO rules_fts(rules_fts) VALUES('rebuild')") await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") status.errors.append(f"Database storage error: {str(e)}") raise async def _generate_embeddings(self, rules: List[RuleData], status: IngestionStatus): """Generate vector embeddings for semantic search""" async with aiosqlite.connect(self.db_path) as db: # Enable sqlite-vss extension await db.enable_load_extension(True) await db.load_extension("vss0") # Clear existing embeddings await db.execute("DELETE FROM vss_rules") # Generate embeddings in batches batch_size = 50 for i in range(0, len(rules), batch_size): batch = rules[i:i + batch_size] # Generate embeddings for batch embedding_tasks = [ self.embedding_service.embed(f"{rule.rule_title} {rule.rule_text}") for rule in batch ] embeddings = await asyncio.gather(*embedding_tasks) # Store embeddings embedding_tuples = [ (rule.rule_id, embedding.tolist()) for rule, embedding in zip(batch, embeddings) ] await db.executemany(""" INSERT INTO vss_rules (rule_id, embedding) VALUES (?, ?) """, embedding_tuples) # Yield control for async processing await asyncio.sleep(0) # MCP tool for PDF ingestion @app.tool("ingest_pdf_rules") async def ingest_pdf_rules(pdf_path: str, version: int = 1) -> Dict: """ Ingest USA Swimming rules from PDF document Args: pdf_path: Path to the USA Swimming rules PDF file version: Version number for this rule set Returns: Dict: Ingestion status and results """ async with PdfRuleIngestionPipeline() as pipeline: status = await pipeline.ingest_pdf(Path(pdf_path), version) return { "status": status.status, "total_pages": status.total_pages, "processed_pages": status.processed_pages, "total_rules": status.total_rules, "processed_rules": status.processed_rules, "errors": status.errors, "warnings": status.warnings, "processing_time": asyncio.get_event_loop().time() - status.start_time } @app.tool("validate_pdf_structure") async def validate_pdf_structure(pdf_path: str) -> Dict: """ Validate PDF structure before ingestion Args: pdf_path: Path to the PDF file to validate Returns: Dict: Validation results and recommendations """ validation_results = { "is_valid": False, "page_count": 0, "has_text": False, "has_structure": False, "estimated_rules": 0, "warnings": [], "recommendations": [] } try: with pdfplumber.open(pdf_path) as pdf: validation_results["page_count"] = len(pdf.pages) # Check first few pages for text content text_content = "" for i, page in enumerate(pdf.pages[:5]): text_content += page.extract_text() or "" validation_results["has_text"] = len(text_content.strip()) > 100 # Check for rule structure patterns rule_pattern = re.compile(r'^\d+(?:\.\d+)*\s+[A-Z]', re.MULTILINE) rule_matches = rule_pattern.findall(text_content) validation_results["estimated_rules"] = len(rule_matches) validation_results["has_structure"] = len(rule_matches) > 0 # Generate recommendations if not validation_results["has_text"]: validation_results["warnings"].append("PDF appears to contain no extractable text") validation_results["recommendations"].append("Ensure PDF is text-based, not scanned images") if not validation_results["has_structure"]: validation_results["warnings"].append("No rule structure detected in sample pages") validation_results["recommendations"].append("Verify this is a USA Swimming rules document") validation_results["is_valid"] = ( validation_results["has_text"] and validation_results["has_structure"] and validation_results["page_count"] > 10 ) except Exception as e: validation_results["warnings"].append(f"PDF validation error: {str(e)}") return validation_results ``` ### 8.2 MCP Server Architecture ```python # Example MCP tools structure with async SQLite and RAG integration from fastmcp import FastMCP <!-- ...existing code... --> ```` The PRD now includes a comprehensive PDF rule ingestion pipeline that: 1. **Extracts structured content** from USA Swimming PDF documents using pdfplumber 2. **Parses rule hierarchies** with regex patterns for rule numbers, sections, and categories 3. **Validates rule data** for completeness and accuracy before storage 4. **Stores rules asynchronously** in SQLite with proper transaction handling 5. **Generates vector embeddings** for semantic search capabilities 6. **Provides status tracking** throughout the ingestion process 7. **Includes validation tools** to verify PDF structure before processing 8. **Handles errors gracefully** with comprehensive error reporting and warnings This pipeline ensures that USA Swimming rules can be reliably extracted from official PDF documents and made available for the swim rules agent's analysis capabilities. Made changes. jimthompson5802: replace the use of sqlite-vss with sqlite-vec GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Replacing sqlite-vss with sqlite-vec throughout the PRD for vector similarity search functionality. ````markdown <!-- ...existing code... --> ### 5.1 Architecture - **Web Framework**: FastAPI for high-performance async API endpoints - **MCP Integration**: FastMCP Python SDK with async RAG-enhanced tools - **Database**: SQLite3 with aiosqlite for async operations + sqlite-vec extension for vector similarity search - **Frontend**: React or Vue.js for interactive UI with async semantic search components - **Transport**: HTTP-based MCP transport with async connection handling - **Async Components**: - AsyncIO event loop for concurrent request handling - Background task queues (Celery/Redis or FastAPI BackgroundTasks) - Async context managers for resource management - Connection pooling for database async operations - **RAG Components**: - Async embedding model interfaces (e.g., sentence-transformers, OpenAI embeddings) - SQLite-vec for vector embeddings and similarity search - Async retrieval system with concurrent similarity ranking - Async context augmentation pipeline with streaming responses ### 5.2 Async-Specific Requirements - **Concurrent Request Handling**: Support 100+ simultaneous async requests with SQLite connection pooling - **Non-blocking Operations**: All database and MCP operations must be async/await compatible using aiosqlite - **Background Task Processing**: Async queue processing for rule updates and embedding generation - **Async Resource Management**: Use AsyncExitStack for proper cleanup of async resources - **Streaming Responses**: Support async streaming for large rule analysis results - **Connection Pooling**: Async SQLite connection pools with configurable limits ### 5.3 RAG-Specific Requirements - **Async Embedding Generation**: Non-blocking embedding creation for new rules using SQLite transactions - **Concurrent Vector Search**: Parallel similarity search across 10,000+ rule embeddings using sqlite-vec - **Async Context Retrieval**: Non-blocking retrieval of top-k (3-5) most relevant rule contexts - **Streaming Response Augmentation**: Async integration of retrieved context into streaming responses - **Async Embedding Quality Validation**: Background validation maintaining cosine similarity > 0.8 <!-- ...existing code... --> ### 5.6 Data Requirements - USA Swimming Official Rules and Regulations sourced from PDF documents - Technical Rules for competitive swimming with full-text search indexes - Administrative regulations with vector embeddings in sqlite-vec tables - Historical rule versions and change logs in versioned SQLite tables - Rule interpretation guidelines with semantic search capabilities ### 5.7 PDF Rule Ingestion Requirements - **PDF Processing**: Automated extraction of structured rule data from USA Swimming PDF documents - **Rule Parsing**: Intelligent parsing of rule numbers, titles, text, and hierarchical relationships - **Data Validation**: Comprehensive validation of extracted rule data for completeness and accuracy - **Version Management**: Track rule document versions and handle incremental updates - **Error Handling**: Robust error handling for PDF parsing failures and data inconsistencies - **Async Processing**: Non-blocking PDF processing with progress tracking and status updates <!-- ...existing code... --> async def _generate_embeddings(self, rules: List[RuleData], status: IngestionStatus): """Generate vector embeddings for semantic search""" async with aiosqlite.connect(self.db_path) as db: # Enable sqlite-vec extension await db.enable_load_extension(True) await db.load_extension("vec0") # Clear existing embeddings await db.execute("DELETE FROM vec_rules") # Generate embeddings in batches batch_size = 50 for i in range(0, len(rules), batch_size): batch = rules[i:i + batch_size] # Generate embeddings for batch embedding_tasks = [ self.embedding_service.embed(f"{rule.rule_title} {rule.rule_text}") for rule in batch ] embeddings = await asyncio.gather(*embedding_tasks) # Store embeddings embedding_tuples = [ (rule.rule_id, embedding.tolist()) for rule, embedding in zip(batch, embeddings) ] await db.executemany(""" INSERT INTO vec_rules (rule_id, embedding) VALUES (?, ?) """, embedding_tuples) # Yield control for async processing await asyncio.sleep(0) # MCP tool for PDF ingestion @app.tool("ingest_pdf_rules") async def ingest_pdf_rules(pdf_path: str, version: int = 1) -> Dict: """ Ingest USA Swimming rules from PDF document Args: pdf_path: Path to the USA Swimming rules PDF file version: Version number for this rule set Returns: Dict: Ingestion status and results """ async with PdfRuleIngestionPipeline() as pipeline: status = await pipeline.ingest_pdf(Path(pdf_path), version) return { "status": status.status, "total_pages": status.total_pages, "processed_pages": status.processed_pages, "total_rules": status.total_rules, "processed_rules": status.processed_rules, "errors": status.errors, "warnings": status.warnings, "processing_time": asyncio.get_event_loop().time() - status.start_time } <!-- ...existing code... --> @app.tool("search_rules") async def search_rules(query: str, category: str = None, use_semantic: bool = True) -> SearchResults: """Search USA Swimming rules by query with optional semantic search using async SQLite operations""" async with aiosqlite.connect(DATABASE_PATH) as db: # Enable sqlite-vec extension for vector search await db.enable_load_extension(True) await db.load_extension("vec0") if use_semantic: # Vector similarity search using sqlite-vec cursor = await db.execute(""" SELECT rule_id, rule_text, distance FROM vec_rules WHERE vec_search(embedding, ?) ORDER BY distance ASC LIMIT 5 """, (query,)) else: # Traditional full-text search cursor = await db.execute(""" SELECT rule_id, rule_text FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank """, (query,)) results = await cursor.fetchall() return SearchResults(results=results) @app.tool("validate_scenario") async def validate_scenario(scenario: str, use_rag: bool = True) -> ViolationAnalysis: """Analyze swimming scenario for rule violations with RAG-enhanced interpretation using async SQLite processing""" async with AsyncExitStack() as stack: # Async resource management with SQLite connections db = await stack.enter_async_context(aiosqlite.connect(DATABASE_PATH)) await db.enable_load_extension(True) await db.load_extension("vec0") # Concurrent processing of multiple rule checks tasks = [ check_stroke_violations(db, scenario), check_timing_violations(db, scenario), check_equipment_violations(db, scenario) ] results = await asyncio.gather(*tasks) return aggregate_analysis(results) <!-- ...existing code... --> @app.tool("semantic_search") async def semantic_search(query: str, top_k: int = 5) -> List[RuleMatch]: """Perform vector-based semantic search across rule embeddings with async SQLite operations""" async with aiosqlite.connect(DATABASE_PATH) as db: await db.enable_load_extension(True) await db.load_extension("vec0") cursor = await db.execute(""" SELECT rule_id, rule_text, rule_number, distance FROM vec_rules WHERE vec_search(embedding, ?) ORDER BY distance ASC LIMIT ? """, (query, top_k)) matches = await cursor.fetchall() return [RuleMatch(rule_id=m[0], text=m[1], number=m[2], similarity=1-m[3]) for m in matches] <!-- ...existing code... --> async def update_rules_async() -> None: """Background async rule update processing with SQLite""" async with aiosqlite.connect(DATABASE_PATH) as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Update rules and regenerate embeddings await db.execute("DELETE FROM rules WHERE version < ?", (new_version,)) await db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?)", new_rules) # Update vector embeddings await db.enable_load_extension(True) await db.load_extension("vec0") await db.execute("DELETE FROM vec_rules") await db.executemany("INSERT INTO vec_rules VALUES (?, ?)", rule_embeddings) await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") raise e ### 8.3 Async SQLite Database Schema Operations - **Async Connection Management**: Connection pooling with aiosqlite for SQLite operations - **Non-blocking Queries**: All database operations use async/await patterns with aiosqlite - **Concurrent Transactions**: SQLite WAL mode for concurrent readers with single writer - **Async Vector Indexing**: Background async indexing for vector embeddings using sqlite-vec - **Streaming Results**: Async generators for large result sets from SQLite queries ### 8.4 Async RAG Pipeline Architecture ```python # Async RAG Pipeline Implementation with SQLite class AsyncSQLiteRAGPipeline: def __init__(self, db_path: str = DATABASE_PATH): self.db_path = db_path self.embedding_service = None self.db_connection = None async def __aenter__(self): """Async context manager entry with SQLite connection""" self.db_connection = await aiosqlite.connect(self.db_path) await self.db_connection.enable_load_extension(True) await self.db_connection.load_extension("vec0") # Configure SQLite for optimal performance await self.db_connection.execute("PRAGMA journal_mode=WAL") await self.db_connection.execute("PRAGMA synchronous=NORMAL") await self.db_connection.execute("PRAGMA cache_size=10000") self.embedding_service = await AsyncEmbeddingService().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" if self.embedding_service: await self.embedding_service.close() if self.db_connection: await self.db_connection.close() async def process_query(self, query: str) -> EnhancedResponse: """Async RAG query processing with concurrent SQLite operations""" # Concurrent embedding generation and similarity search embedding_task = asyncio.create_task(self.embedding_service.embed(query)) embedding = await embedding_task # Concurrent context retrieval using SQLite similarity_task = asyncio.create_task(self._similarity_search(embedding)) rule_task = asyncio.create_task(self._get_related_rules(query)) similar_contexts, related_rules = await asyncio.gather(similarity_task, rule_task) # Async response augmentation return await self.augment_response(query, similar_contexts, related_rules) async def _similarity_search(self, embedding: List[float]) -> List[RuleContext]: """Perform vector similarity search using sqlite-vec""" cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, distance FROM vec_rules WHERE vec_search(embedding, ?) ORDER BY distance ASC LIMIT 5 """, (embedding,)) results = await cursor.fetchall() return [RuleContext(id=r[0], text=r[1], similarity=1-r[2]) for r in results] async def _get_related_rules(self, query: str) -> List[Rule]: """Get related rules using full-text search""" cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, rule_number FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank LIMIT 10 """, (query,)) results = await cursor.fetchall() return [Rule(id=r[0], text=r[1], number=r[2]) for r in results] ``` ### 8.6 SQLite Database Schema ```sql -- Core rules table with full-text search CREATE TABLE rules ( rule_id TEXT PRIMARY KEY, rule_number TEXT NOT NULL, rule_text TEXT NOT NULL, category TEXT NOT NULL, effective_date DATE, version INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Full-text search virtual table CREATE VIRTUAL TABLE rules_fts USING fts5( rule_id UNINDEXED, rule_number, rule_text, category, content='rules', content_rowid='rowid' ); -- Vector embeddings table using sqlite-vec CREATE VIRTUAL TABLE vec_rules USING vec0( embedding float[1536], -- Dimension for OpenAI embeddings rule_id TEXT PRIMARY KEY ); -- Rule relationships and cross-references CREATE TABLE rule_relationships ( relationship_id INTEGER PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), related_rule_id TEXT REFERENCES rules(rule_id), relationship_type TEXT NOT NULL, strength REAL DEFAULT 1.0 ); -- Historical rule versions CREATE TABLE rule_history ( history_id INTEGER PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), previous_text TEXT, change_description TEXT, changed_date DATE, version INTEGER ); -- Violation patterns and triggers CREATE TABLE violations ( violation_id TEXT PRIMARY KEY, rule_id TEXT REFERENCES rules(rule_id), violation_type TEXT NOT NULL, severity TEXT NOT NULL, penalty_description TEXT, trigger_patterns TEXT -- JSON array of pattern strings ); -- Analyzed scenarios for learning CREATE TABLE scenarios ( scenario_id TEXT PRIMARY KEY, scenario_text TEXT NOT NULL, analysis_result TEXT, -- JSON result applicable_rules TEXT, -- JSON array of rule_ids created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, embedding BLOB -- Serialized embedding vector ); -- Performance indexes CREATE INDEX idx_rules_category ON rules(category); CREATE INDEX idx_rules_version ON rules(version); CREATE INDEX idx_violations_type ON violations(violation_type); CREATE INDEX idx_scenarios_created ON scenarios(created_at); ``` <!-- ...existing code... --> ```` The PRD has been updated to use sqlite-vec instead of sqlite-vss for vector similarity search, including: 1. **Architecture changes**: Updated to use sqlite-vec extension (`vec0`) instead of sqlite-vss (`vss0`) 2. **Database schema**: Changed virtual table from `vss_rules` to `vec_rules` with sqlite-vec syntax 3. **Query syntax**: Updated from `vss_search()` to `vec_search()` function calls 4. **Extension loading**: Changed from `load_extension("vss0")` to `load_extension("vec0")` 5. **Vector operations**: Updated all vector search operations to use sqlite-vec API This change provides a more lightweight and efficient vector search solution while maintaining all the RAG functionality. Made changes. jimthompson5802: ensure any code shown in the prd is python GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Fixing syntax errors in the Python code examples throughout the PRD to ensure proper Python syntax. ````markdown <!-- ...existing code... --> try: # Extract text from PDF extracted_data = await self._extract_pdf_content(pdf_path, status) # Parse rules from extracted text parsed_rules = await self._parse_rules(extracted_data, status) # Validate parsed rules validated_rules = await self._validate_rules(parsed_rules, status) # Store rules in database await self._store_rules(validated_rules, version, status) # Generate embeddings await self._generate_embeddings(validated_rules, status) status.status = 'completed' except Exception as e: status.errors.append(f"Ingestion failed: {str(e)}") status.status = 'failed' raise return status async def _extract_pdf_content(self, pdf_path: Path, status: IngestionStatus) -> List[Dict]: """Extract structured content from PDF pages""" extracted_pages = [] with pdfplumber.open(pdf_path) as pdf: status.total_pages = len(pdf.pages) for page_num, page in enumerate(pdf.pages): try: # Extract text with layout preservation text = page.extract_text(layout=True) # Extract tables if present tables = page.extract_tables() # Extract metadata page_data = { 'page_number': page_num + 1, 'text': text, 'tables': tables, 'bbox': page.bbox, 'width': page.width, 'height': page.height } extracted_pages.append(page_data) status.processed_pages += 1 # Yield control for async processing if page_num % 10 == 0: await asyncio.sleep(0) except Exception as e: status.warnings.append(f"Error processing page {page_num + 1}: {str(e)}") return extracted_pages async def _parse_rules(self, extracted_pages: List[Dict], status: IngestionStatus) -> List[RuleData]: """Parse rules from extracted PDF content""" rules = [] current_section = "" current_category = "" effective_date = None for page_data in extracted_pages: text_lines = page_data['text'].split('\n') for line_num, line in enumerate(text_lines): line = line.strip() if not line: continue try: # Check for section headers section_match = self.rule_patterns['section_header'].match(line) if section_match: current_section = section_match.group(1) current_category = section_match.group(2).strip() continue # Check for effective date date_match = self.rule_patterns['effective_date'].search(line) if date_match: effective_date = date_match.group(1).strip() continue # Check for rule number and content rule_match = self.rule_patterns['rule_number'].match(line) if rule_match: rule_number = rule_match.group(1) rule_title = rule_match.group(2).strip() # Extract rule text (may span multiple lines) rule_text = await self._extract_rule_text( text_lines, line_num, page_data['page_number'] ) rule_data = RuleData( rule_id=f"rule_{rule_number.replace('.', '_')}", rule_number=rule_number, rule_title=rule_title, rule_text=rule_text, category=current_category, section=current_section, effective_date=effective_date, version=1 ) rules.append(rule_data) status.total_rules += 1 except Exception as e: status.warnings.append( f"Error parsing line {line_num} on page {page_data['page_number']}: {str(e)}" ) # Yield control for async processing await asyncio.sleep(0) return rules async def _extract_rule_text(self, text_lines: List[str], start_line: int, page_number: int) -> str: """Extract complete rule text that may span multiple lines""" rule_text_lines = [] # Start from the line after the rule number for i in range(start_line + 1, len(text_lines)): line = text_lines[i].strip() # Stop if we hit another rule number or section header if (self.rule_patterns['rule_number'].match(line) or self.rule_patterns['section_header'].match(line) or line.startswith('ARTICLE')): break # Skip empty lines at the beginning if not rule_text_lines and not line: continue rule_text_lines.append(line) return ' '.join(rule_text_lines).strip() async def _validate_rules(self, rules: List[RuleData], status: IngestionStatus) -> List[RuleData]: """Validate parsed rules for completeness and accuracy""" validated_rules = [] for rule in rules: try: # Validate rule number format if not re.match(r'^\d+(?:\.\d+)*$', rule.rule_number): status.warnings.append(f"Invalid rule number format: {rule.rule_number}") continue # Validate required fields if not rule.rule_text or len(rule.rule_text.strip()) < 10: status.warnings.append(f"Rule {rule.rule_number} has insufficient text content") continue # Validate rule title if not rule.rule_title: status.warnings.append(f"Rule {rule.rule_number} missing title") continue validated_rules.append(rule) status.processed_rules += 1 except Exception as e: status.errors.append(f"Validation error for rule {rule.rule_number}: {str(e)}") return validated_rules async def _store_rules(self, rules: List[RuleData], version: int, status: IngestionStatus): """Store validated rules in SQLite database""" async with aiosqlite.connect(self.db_path) as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Clear existing rules for this version await db.execute("DELETE FROM rules WHERE version = ?", (version,)) # Insert new rules rule_tuples = [ ( rule.rule_id, rule.rule_number, rule.rule_text, rule.category, rule.effective_date, version, rule.rule_title, rule.section ) for rule in rules ] await db.executemany(""" INSERT INTO rules (rule_id, rule_number, rule_text, category, effective_date, version, rule_title, section) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, rule_tuples) # Update full-text search index await db.execute("INSERT INTO rules_fts(rules_fts) VALUES('rebuild')") await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") status.errors.append(f"Database storage error: {str(e)}") raise # ...existing code... # MCP tool for PDF ingestion @app.tool("ingest_pdf_rules") async def ingest_pdf_rules(pdf_path: str, version: int = 1) -> Dict: """ Ingest USA Swimming rules from PDF document Args: pdf_path: Path to the USA Swimming rules PDF file version: Version number for this rule set Returns: Dict: Ingestion status and results """ async with PdfRuleIngestionPipeline() as pipeline: status = await pipeline.ingest_pdf(Path(pdf_path), version) return { "status": status.status, "total_pages": status.total_pages, "processed_pages": status.processed_pages, "total_rules": status.total_rules, "processed_rules": status.processed_rules, "errors": status.errors, "warnings": status.warnings, "processing_time": asyncio.get_event_loop().time() - status.start_time } @app.tool("validate_pdf_structure") async def validate_pdf_structure(pdf_path: str) -> Dict: """ Validate PDF structure before ingestion Args: pdf_path: Path to the PDF file to validate Returns: Dict: Validation results and recommendations """ validation_results = { "is_valid": False, "page_count": 0, "has_text": False, "has_structure": False, "estimated_rules": 0, "warnings": [], "recommendations": [] } try: with pdfplumber.open(pdf_path) as pdf: validation_results["page_count"] = len(pdf.pages) # Check first few pages for text content text_content = "" for i, page in enumerate(pdf.pages[:5]): text_content += page.extract_text() or "" validation_results["has_text"] = len(text_content.strip()) > 100 # Check for rule structure patterns rule_pattern = re.compile(r'^\d+(?:\.\d+)*\s+[A-Z]', re.MULTILINE) rule_matches = rule_pattern.findall(text_content) validation_results["estimated_rules"] = len(rule_matches) validation_results["has_structure"] = len(rule_matches) > 0 # Generate recommendations if not validation_results["has_text"]: validation_results["warnings"].append("PDF appears to contain no extractable text") validation_results["recommendations"].append("Ensure PDF is text-based, not scanned images") if not validation_results["has_structure"]: validation_results["warnings"].append("No rule structure detected in sample pages") validation_results["recommendations"].append("Verify this is a USA Swimming rules document") validation_results["is_valid"] = ( validation_results["has_text"] and validation_results["has_structure"] and validation_results["page_count"] > 10 ) except Exception as e: validation_results["warnings"].append(f"PDF validation error: {str(e)}") return validation_results # ...existing code... async def update_rules_async() -> None: """Background async rule update processing with SQLite""" async with aiosqlite.connect(DATABASE_PATH) as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Update rules and regenerate embeddings await db.execute("DELETE FROM rules WHERE version < ?", (new_version,)) await db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?)", new_rules) # Update vector embeddings await db.enable_load_extension(True) await db.load_extension("vec0") await db.execute("DELETE FROM vec_rules") await db.executemany("INSERT INTO vec_rules VALUES (?, ?)", rule_embeddings) await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") raise e # ...existing code... ```` Fixed all Python syntax errors in the PRD code examples: 1. **Removed JavaScript-style braces**: Changed `try {` to `try:` and `} except` to `except` 2. **Fixed exception handling**: Properly formatted try/except blocks with colons 3. **Corrected indentation**: Ensured proper Python indentation throughout 4. **Fixed method calls**: Removed unnecessary braces and formatting issues 5. **Maintained async/await patterns**: Kept all async functionality intact 6. **Preserved functionality**: All code maintains the same logic and behavior All code in the PRD now follows proper Python syntax and PEP 8 conventions. Made changes. jimthompson5802: update the prd to specify the programming language is python GitHub Copilot: ### [product_requirements_doc.md](file:///Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md) Adding Python language specification throughout the PRD technical specifications and fixing remaining syntax errors. ````markdown <!-- ...existing code... --> ## 5. Technical Requirements ### 5.1 Architecture - **Programming Language**: Python 3.11+ with asyncio for concurrent operations - **Web Framework**: FastAPI for high-performance async API endpoints - **MCP Integration**: FastMCP Python SDK with async RAG-enhanced tools - **Database**: SQLite3 with aiosqlite for async operations + sqlite-vec extension for vector similarity search - **Frontend**: React or Vue.js for interactive UI with async semantic search components - **Transport**: HTTP-based MCP transport with async connection handling - **Async Components**: - AsyncIO event loop for concurrent request handling - Background task queues (Celery/Redis or FastAPI BackgroundTasks) - Async context managers for resource management - Connection pooling for database async operations - **RAG Components**: - Async embedding model interfaces (e.g., sentence-transformers, OpenAI embeddings) - SQLite-vec for vector embeddings and similarity search - Async retrieval system with concurrent similarity ranking - Async context augmentation pipeline with streaming responses ### 5.2 Python-Specific Requirements - **Python Version**: Python 3.11 or higher for optimal async performance - **Type Hints**: Complete type annotations following PEP 484 and PEP 585 - **Code Style**: PEP 8 compliance with 120-character line length limit - **Documentation**: Google-style docstrings for all classes and functions - **Error Handling**: Comprehensive try/except blocks with descriptive error messages - **Async Programming**: Proper use of async/await patterns and AsyncExitStack for resource management <!-- ...existing code... --> ### 5.7 PDF Rule Ingestion Requirements - **PDF Processing**: Python-based automated extraction using pdfplumber library - **Rule Parsing**: Intelligent parsing using Python regex patterns for rule structure - **Data Validation**: Python validation functions for rule completeness and accuracy - **Version Management**: Python-based tracking of rule document versions - **Error Handling**: Robust Python exception handling for PDF parsing failures - **Async Processing**: Python asyncio-based non-blocking PDF processing ## 8. Technical Specifications ### 8.1 PDF Rule Ingestion Pipeline (Python Implementation) ```python # filepath: /Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md # PDF Rule Ingestion Pipeline with async processing from fastmcp import FastMCP from typing import List, Dict, Optional, Tuple import asyncio import aiosqlite import pdfplumber import re import json from dataclasses import dataclass from pathlib import Path from contextlib import AsyncExitStack @dataclass class RuleData: """Data class for parsed rule information Attributes: rule_id: Unique identifier for the rule rule_number: Official rule number (e.g., "101.2.2") rule_title: Title or heading of the rule rule_text: Complete text content of the rule category: Rule category (e.g., "Stroke Technique") section: Section number within the rule document subsection: Optional subsection identifier effective_date: Date when rule becomes effective version: Version number of the rule set """ rule_id: str rule_number: str rule_title: str rule_text: str category: str section: str subsection: Optional[str] = None effective_date: Optional[str] = None version: int = 1 @dataclass class IngestionStatus: """Status tracking for PDF ingestion process Attributes: total_pages: Total number of pages in PDF processed_pages: Number of pages successfully processed total_rules: Total number of rules found processed_rules: Number of rules successfully processed errors: List of error messages encountered warnings: List of warning messages start_time: Processing start time status: Current processing status """ total_pages: int processed_pages: int total_rules: int processed_rules: int errors: List[str] warnings: List[str] start_time: float status: str # 'processing', 'completed', 'failed' class PdfRuleIngestionPipeline: """Async PDF rule ingestion pipeline for USA Swimming rules This class handles the complete pipeline for extracting, parsing, validating, and storing USA Swimming rules from PDF documents into a SQLite database with vector embeddings for semantic search. """ def __init__(self, db_path: str = "swim_rules.db") -> None: """Initialize the PDF ingestion pipeline Args: db_path: Path to the SQLite database file """ self.db_path = db_path self.embedding_service = None self.rule_patterns = { 'rule_number': re.compile(r'^(\d+(?:\.\d+)*)\s+([A-Z][^.]*?)(?:\s*—\s*(.*))?$'), 'section_header': re.compile(r'^ARTICLE\s+(\d+)\s*[—-]\s*(.+)$'), 'subsection': re.compile(r'^(\d+\.\d+)\s+(.+)$'), 'effective_date': re.compile(r'Effective\s+Date:\s*(.+)', re.IGNORECASE) } async def __aenter__(self): """Async context manager entry""" self.embedding_service = await AsyncEmbeddingService().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" if self.embedding_service: await self.embedding_service.close() <!-- ...existing code... --> # MCP tool for PDF ingestion app = FastMCP("swim-rules-agent") @app.tool("ingest_pdf_rules") async def ingest_pdf_rules(pdf_path: str, version: int = 1) -> Dict[str, any]: """ Ingest USA Swimming rules from PDF document Args: pdf_path: Path to the USA Swimming rules PDF file version: Version number for this rule set Returns: Dict: Ingestion status and results including processing metrics """ async with PdfRuleIngestionPipeline() as pipeline: status = await pipeline.ingest_pdf(Path(pdf_path), version) return { "status": status.status, "total_pages": status.total_pages, "processed_pages": status.processed_pages, "total_rules": status.total_rules, "processed_rules": status.processed_rules, "errors": status.errors, "warnings": status.warnings, "processing_time": asyncio.get_event_loop().time() - status.start_time } @app.tool("validate_pdf_structure") async def validate_pdf_structure(pdf_path: str) -> Dict[str, any]: """ Validate PDF structure before ingestion Args: pdf_path: Path to the PDF file to validate Returns: Dict: Validation results and recommendations """ validation_results = { "is_valid": False, "page_count": 0, "has_text": False, "has_structure": False, "estimated_rules": 0, "warnings": [], "recommendations": [] } try: with pdfplumber.open(pdf_path) as pdf: validation_results["page_count"] = len(pdf.pages) # Check first few pages for text content text_content = "" for i, page in enumerate(pdf.pages[:5]): text_content += page.extract_text() or "" validation_results["has_text"] = len(text_content.strip()) > 100 # Check for rule structure patterns rule_pattern = re.compile(r'^\d+(?:\.\d+)*\s+[A-Z]', re.MULTILINE) rule_matches = rule_pattern.findall(text_content) validation_results["estimated_rules"] = len(rule_matches) validation_results["has_structure"] = len(rule_matches) > 0 # Generate recommendations if not validation_results["has_text"]: validation_results["warnings"].append("PDF appears to contain no extractable text") validation_results["recommendations"].append("Ensure PDF is text-based, not scanned images") if not validation_results["has_structure"]: validation_results["warnings"].append("No rule structure detected in sample pages") validation_results["recommendations"].append("Verify this is a USA Swimming rules document") validation_results["is_valid"] = ( validation_results["has_text"] and validation_results["has_structure"] and validation_results["page_count"] > 10 ) except Exception as e: validation_results["warnings"].append(f"PDF validation error: {str(e)}") return validation_results # Additional MCP tools with proper Python typing @app.tool("search_rules") async def search_rules(query: str, category: Optional[str] = None, use_semantic: bool = True) -> Dict[str, any]: """Search USA Swimming rules by query with optional semantic search using async SQLite operations Args: query: Search query string category: Optional category filter use_semantic: Whether to use semantic vector search Returns: Dict: Search results with rule matches """ async with aiosqlite.connect("swim_rules.db") as db: # Enable sqlite-vec extension for vector search await db.enable_load_extension(True) await db.load_extension("vec0") if use_semantic: # Vector similarity search using sqlite-vec cursor = await db.execute(""" SELECT rule_id, rule_text, distance FROM vec_rules WHERE vec_search(embedding, ?) ORDER BY distance ASC LIMIT 5 """, (query,)) else: # Traditional full-text search cursor = await db.execute(""" SELECT rule_id, rule_text FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank """, (query,)) results = await cursor.fetchall() return {"results": results, "count": len(results)} <!-- ...existing code... --> async def update_rules_async() -> None: """Background async rule update processing with SQLite This function handles the complete rule update process including database transactions, embedding regeneration, and error handling. """ async with aiosqlite.connect("swim_rules.db") as db: # Begin transaction for atomic updates await db.execute("BEGIN TRANSACTION") try: # Update rules and regenerate embeddings await db.execute("DELETE FROM rules WHERE version < ?", (2,)) # Example version # await db.executemany("INSERT INTO rules VALUES (?, ?, ?, ?)", new_rules) # Update vector embeddings await db.enable_load_extension(True) await db.load_extension("vec0") await db.execute("DELETE FROM vec_rules") # await db.executemany("INSERT INTO vec_rules VALUES (?, ?)", rule_embeddings) await db.execute("COMMIT") except Exception as e: await db.execute("ROLLBACK") raise e ``` ### 8.2 Python Async RAG Pipeline Architecture ```python # filepath: /Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md # Async RAG Pipeline Implementation with SQLite using Python from typing import List, Optional, Dict, Any import asyncio import aiosqlite from contextlib import AsyncExitStack class AsyncSQLiteRAGPipeline: """Python-based async RAG pipeline for swim rules analysis This class implements a Retrieval-Augmented Generation pipeline using SQLite with vector extensions for semantic search and context retrieval in swimming rule analysis. """ def __init__(self, db_path: str = "swim_rules.db") -> None: """Initialize the RAG pipeline Args: db_path: Path to the SQLite database file """ self.db_path = db_path self.embedding_service: Optional[Any] = None self.db_connection: Optional[aiosqlite.Connection] = None async def __aenter__(self): """Async context manager entry with SQLite connection""" self.db_connection = await aiosqlite.connect(self.db_path) await self.db_connection.enable_load_extension(True) await self.db_connection.load_extension("vec0") # Configure SQLite for optimal performance await self.db_connection.execute("PRAGMA journal_mode=WAL") await self.db_connection.execute("PRAGMA synchronous=NORMAL") await self.db_connection.execute("PRAGMA cache_size=10000") self.embedding_service = await AsyncEmbeddingService().connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager cleanup""" if self.embedding_service: await self.embedding_service.close() if self.db_connection: await self.db_connection.close() async def process_query(self, query: str) -> Dict[str, Any]: """Async RAG query processing with concurrent SQLite operations Args: query: User query for rule analysis Returns: Dict: Enhanced response with retrieved context """ # Concurrent embedding generation and similarity search embedding_task = asyncio.create_task(self.embedding_service.embed(query)) embedding = await embedding_task # Concurrent context retrieval using SQLite similarity_task = asyncio.create_task(self._similarity_search(embedding)) rule_task = asyncio.create_task(self._get_related_rules(query)) similar_contexts, related_rules = await asyncio.gather(similarity_task, rule_task) # Async response augmentation return await self.augment_response(query, similar_contexts, related_rules) async def _similarity_search(self, embedding: List[float]) -> List[Dict[str, Any]]: """Perform vector similarity search using sqlite-vec Args: embedding: Vector embedding for similarity search Returns: List of rule contexts with similarity scores """ cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, distance FROM vec_rules WHERE vec_search(embedding, ?) ORDER BY distance ASC LIMIT 5 """, (embedding,)) results = await cursor.fetchall() return [{"id": r[0], "text": r[1], "similarity": 1-r[2]} for r in results] async def _get_related_rules(self, query: str) -> List[Dict[str, Any]]: """Get related rules using full-text search Args: query: Search query string Returns: List of related rules from full-text search """ cursor = await self.db_connection.execute(""" SELECT rule_id, rule_text, rule_number FROM rules_fts WHERE rules_fts MATCH ? ORDER BY rank LIMIT 10 """, (query,)) results = await cursor.fetchall() return [{"id": r[0], "text": r[1], "number": r[2]} for r in results] ``` ### 8.3 Python SQLite Database Schema ```sql -- filepath: /Users/jim/Desktop/modelcontextprotocol/mcp_testlab/mcp-swimrules/product_requirements_doc.md -- SQLite database schema for Python-based swim rules application -- Core rules table with full-text search CREATE TABLE rules ( rule_id TEXT PRIMARY KEY, rule_number TEXT NOT NULL, rule_text TEXT NOT NULL, category TEXT NOT NULL, effective_date DATE, version INTEGER DEFAULT 1, rule_title TEXT, section TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Full-text search virtual table CREATE VIRTUAL TABLE rules_fts USING fts5( rule_id UNINDEXED, rule_number, rule_text, category, content='rules', content_rowid='rowid' ); -- Vector embeddings table using sqlite-vec CREATE VIRTUAL TABLE vec_rules USING vec0( embedding float[1536], -- Dimension for OpenAI embeddings rule_id TEXT PRIMARY KEY ); <!-- ...existing code... --> -- Performance indexes CREATE INDEX idx_rules_category ON rules(category); CREATE INDEX idx_rules_version ON rules(version); CREATE INDEX idx_violations_type ON violations(violation_type); CREATE INDEX idx_scenarios_created ON scenarios(created_at); ``` ### 8.4 Python Development Environment Requirements - **Python Version**: 3.11+ with typing support - **Required Packages**: - `fastmcp`: MCP server and client functionality - `fastapi`: Web framework for REST API - `aiosqlite`: Async SQLite database operations - `pdfplumber`: PDF text extraction - `sqlite-vec`: Vector similarity search extension - `asyncio`: Built-in async/await support - **Development Tools**: - `pytest`: Testing framework - `black`: Code formatting - `mypy`: Static type checking - `ruff`: Fast Python linter <!-- ...existing code... --> ```` The PRD now clearly specifies Python as the programming language throughout, with: 1. **Language specification** in the architecture section 2. **Python-specific requirements** including version, type hints, and coding standards 3. **Proper Python typing** in all code examples with type annotations 4. **Google-style docstrings** for all classes and functions 5. **PEP 8 compliance** mentioned in requirements 6. **Python package requirements** clearly listed 7. **Fixed syntax errors** in the remaining code blocks 8. **Consistent Python conventions** throughout all examples Made changes.