WIP

Author: kwojcik-blockether

src/blockether_catalyst/integrations/agno/AgnoOsASGIModule.py

@@ -100,7 +100,6 @@ class MCPConfig(BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    workflow: "Workflow" = Field(..., description="The MCP workflow to use")
     name: str = Field(description="Name of the MCP application")
     tools: List[Tool] = Field(default_factory=list, description="List of tools to register with MCP")
     resources: List[Resource] = Field(default_factory=list, description="List of resources for MCP")
@@ -191,6 +190,8 @@ def mount(self, app: FastAPI, router: APIRouter) -> None:
             for tool in self.mcp.tools:
                 mcp.add_tool(tool)
 
+            app.mount("/mcp", mcp.http_app("/", stateless_http=True, transport="sse"))
+
         if not self.chat or not self.chat.assistant:
             raise ValueError("Chat configuration or assistant is not properly set")
 
@@ -232,13 +233,13 @@ async def chat_interface(request: Request) -> HTMLResponse:
 
             if self.workflows:
                 session_type = "workflow"
-                executor_id = self.workflows[0].id if hasattr(self.workflows[0], 'id') else ""
+                executor_id = self.workflows[0].id if hasattr(self.workflows[0], "id") else ""
             elif self.agents:
                 session_type = "agent"
-                executor_id = self.agents[0].id if hasattr(self.agents[0], 'id') else ""
+                executor_id = self.agents[0].id if hasattr(self.agents[0], "id") else ""
             elif self.teams:
                 session_type = "team"
-                executor_id = self.teams[0].id if hasattr(self.teams[0], 'id') else ""
+                executor_id = self.teams[0].id if hasattr(self.teams[0], "id") else ""
 
             # Create the response using render_template which returns a TemplateResponse
             response = self.render_template(
@@ -744,7 +745,7 @@ async def executor_runs(request: Request) -> Dict[str, Any]:
                 return {
                     "error": "Missing required field: 'message'",
                     "status": "error",
-                    "code": 400
+                    "code": 400,
                 }
 
             # Get the runner from the assistant config
@@ -807,64 +808,93 @@ async def executor_runs(request: Request) -> Dict[str, Any]:
                 }
 
         @router_os.get("/sessions/{session_id}/runs", include_in_schema=False)
-        async def get_session_runs(session_id: str) -> List[Dict[str, Any]]:
+        async def get_session_runs(session_id: str, request: Request) -> List[Dict[str, Any]]:
             """Get all runs for a session."""
             try:
                 # Get the runner from the assistant config
                 runner = self.chat.assistant.runner
-                
+
                 # Check if the runner has a database
-                if not hasattr(runner, 'db') or runner.db is None:
+                if not hasattr(runner, "db") or runner.db is None:
                     logger.debug(f"No database configured for runner, returning empty list for session {session_id}")
                     return []
-                
+
+                # Get session type from query params if provided
+                session_type = request.query_params.get("type")
+
                 # Get the session from the database - try different session types
                 session = None
-                for session_type in ["workflow", "agent", "team"]:
+                if session_type:
+                    # If type is specified, only try that type
                     try:
                         session = runner.db.get_session(session_id, session_type)
-                        if session:
-                            break
                     except Exception:
-                        continue
+                        pass
+                else:
+                    # Otherwise try all types
+                    for st in ["agent", "workflow", "team"]:
+                        try:
+                            session = runner.db.get_session(session_id, st)
+                            if session:
+                                break
+                        except Exception:
+                            continue
 
                 if not session:
                     logger.debug(f"No session found for {session_id}")
                     return []
-                
+
                 # Extract runs from the session
                 runs = []
-                if hasattr(session, 'agent_runs'):
+                if hasattr(session, "agent_runs") and session.agent_runs:
                     # For agent sessions
                     for run in session.agent_runs:
-                        runs.append({
-                            "run_id": run.run_id,
-                            "session_id": session_id,
-                            "input": run.user_message,
-                            "output": run.response,
-                            "created_at": run.created_at.isoformat() if hasattr(run.created_at, 'isoformat') else str(run.created_at),
-                        })
-                elif hasattr(session, 'workflow_runs'):
+                        runs.append(
+                            {
+                                "run_id": run.run_id,
+                                "session_id": session_id,
+                                "input": getattr(run, "user_message", getattr(run, "input", "")),
+                                "output": getattr(run, "response", getattr(run, "output", "")),
+                                "created_at": (
+                                    run.created_at.isoformat()
+                                    if hasattr(run.created_at, "isoformat")
+                                    else str(run.created_at)
+                                ),
+                            }
+                        )
+                elif hasattr(session, "workflow_runs") and session.workflow_runs:
                     # For workflow sessions
                     for run in session.workflow_runs:
-                        runs.append({
-                            "run_id": run.run_id,
-                            "session_id": session_id,
-                            "input": run.input,
-                            "output": run.output,
-                            "created_at": run.created_at.isoformat() if hasattr(run.created_at, 'isoformat') else str(run.created_at),
-                        })
-                elif hasattr(session, 'team_runs'):
+                        runs.append(
+                            {
+                                "run_id": run.run_id,
+                                "session_id": session_id,
+                                "input": getattr(run, "input", getattr(run, "user_message", "")),
+                                "output": getattr(run, "output", getattr(run, "response", "")),
+                                "created_at": (
+                                    run.created_at.isoformat()
+                                    if hasattr(run.created_at, "isoformat")
+                                    else str(run.created_at)
+                                ),
+                            }
+                        )
+                elif hasattr(session, "team_runs") and session.team_runs:
                     # For team sessions
                     for run in session.team_runs:
-                        runs.append({
-                            "run_id": run.run_id,
-                            "session_id": session_id,
-                            "input": run.input,
-                            "output": run.output,
-                            "created_at": run.created_at.isoformat() if hasattr(run.created_at, 'isoformat') else str(run.created_at),
-                        })
-                
+                        runs.append(
+                            {
+                                "run_id": run.run_id,
+                                "session_id": session_id,
+                                "input": getattr(run, "input", getattr(run, "user_message", "")),
+                                "output": getattr(run, "output", getattr(run, "response", "")),
+                                "created_at": (
+                                    run.created_at.isoformat()
+                                    if hasattr(run.created_at, "isoformat")
+                                    else str(run.created_at)
+                                ),
+                            }
+                        )
+
                 logger.debug(f"Retrieved {len(runs)} runs for session {session_id}")
                 return runs
 

src/blockether_catalyst/knowledge/KnowledgeExtractionCore.py

@@ -1060,12 +1060,10 @@ async def classify_single_chunk(chunk: KnowledgeChunk) -> KnowledgeChunk:
                 """Classify a single chunk's semantic type - MANDATORY."""
                 # Handle empty or whitespace-only chunks
                 if not chunk.text or not chunk.text.strip():
-                    logger.warning(
-                        f"Chunk {chunk.index} is empty, assigning 'general' semantic type"
-                    )
+                    logger.warning(f"Chunk {chunk.index} is empty, assigning 'general' semantic type")
                     chunk.semantic_types = ["general"]
                     return chunk
-                
+
                 result = await self.calls.chunk_content_classification_call.execute(
                     chunk_text=chunk.text,
                     document_name=doc_result.document_filename,

src/blockether_catalyst/knowledge/KnowledgeSearchCore.py

@@ -24,6 +24,7 @@
     ImageInfo,
     KnowledgeSearchResult,
     LinkedKnowledge,
+    LinkedTermInfo,
     NormalizedSearchResult,
     SearchResult,
     SearchResultMetadata,
@@ -314,7 +315,12 @@ def search(
         # Enhance search results with term analysis and boosting
         # The function now handles efficient top-k selection internally
         enhanced_results = self._enhance_search_results(
-            search_results, query, top_terms, max_depth, max_cooccurrences, k * 2  # Get more for filtering
+            search_results,
+            query,
+            top_terms,
+            max_depth,
+            max_cooccurrences,
+            k * 2,  # Get more for filtering
         )
 
         # Filter enhanced results based on acronym/keyword matches
@@ -765,12 +771,13 @@ def _convert_to_term_info(self, term: TermWithLinks, link_limit: int) -> Optiona
             linked_term = self._resolve_term(link.link_to)
             if linked_term:
                 linked_terms.append(
-                    TermInfo(
+                    LinkedTermInfo(
                         term=linked_term.term,
                         meaning=linked_term.meaning[:600] if linked_term.meaning else None,
                         term_type=linked_term.type,
                         link_score=link.score,
                         total_times_occurred_in_knowledgebase=linked_term.total,
+                        linked_terms=[],  # Usually empty to prevent deep nesting
                     )
                 )
 
@@ -865,9 +872,7 @@ def get_extraction_details(self) -> Dict[str, Any]:
         if not self._linked_knowledge:
             return {}
 
-        return self._linked_knowledge.get_extraction_details(
-            base_url=self._resources_base_url or ""
-        )
+        return self._linked_knowledge.get_extraction_details(base_url=self._resources_base_url or "")
 
     def persist(self, path: Optional[Union[str, Path]] = None) -> None:
         """

src/blockether_catalyst/knowledge/KnowledgeTypes.py

@@ -566,9 +566,8 @@ class TermInfo(TypedPydanticModel):
     term: str = Field(description="The term text")
     meaning: Optional[str] = Field(default=None, description="Meaning of the term")
     term_type: Optional[str] = Field(default=None, description="Type of term (e.g., acronym, keyword)")
-    link_score: Optional[float] = Field(default=None, description="Score of the link to this term")
     total_times_occurred_in_knowledgebase: Optional[int] = Field(default=None, description="Total occurrences")
-    linked_terms: List["TermInfo"] = Field(default_factory=list, description="Linked terms")
+    linked_terms: List["LinkedTermInfo"] = Field(default_factory=list, description="Terms linked from this term")
 
     def markdown(self) -> str:
         """Generate markdown for a term."""
@@ -598,6 +597,15 @@ def markdown(self) -> str:
         return "\n".join(lines)
 
 
+class LinkedTermInfo(TermInfo):
+    """Information about a term that is linked from another term (extends TermInfo with link_score)."""
+
+    link_score: float = Field(description="Score of the link to this term (strength of relationship, 0.0-1.0)")
+    linked_terms: List["LinkedTermInfo"] = Field(
+        default_factory=list, description="Nested linked terms (usually empty)"
+    )
+
+
 class ImageInfo(TypedPydanticModel):
     """Information about an image."""
 

src/blockether_catalyst/knowledge/PDFKnowledgeExtractor.py

@@ -139,7 +139,12 @@ def _process_page(
         base_text = page_without_tables.extract_text(**self._text_extraction_settings.model_dump()) or ""
 
         # Extract image
-        images = self._extract_images_from_page(page, context=base_text, current_document=current_document, current_document_path=current_document_path)
+        images = self._extract_images_from_page(
+            page,
+            context=base_text,
+            current_document=current_document,
+            current_document_path=current_document_path,
+        )
 
         # Fix hyphenated line breaks immediately after extraction
         base_text = self._fix_hyphenated_line_breaks(base_text)
@@ -256,9 +261,7 @@ def _extract_tables_from_page(
                     tables.append(pdf_table)
 
             except Exception as e:
-                self._logger.warning(
-                    f"[{current_document}] Error extracting table on page {page.page_number}: {e}"
-                )
+                self._logger.warning(f"[{current_document}] Error extracting table on page {page.page_number}: {e}")
 
         return tables