add timeouts to tools

Author: Lekssays

src/tools/code_browsing_tools.py

@@ -51,7 +51,8 @@ def register_code_browsing_tools(mcp, services: dict):
 
 Examples:
     list_methods(codebase_hash="abc", name_pattern=".*auth.*")
-    list_methods(codebase_hash="abc", callee_pattern="memcpy")"""
+    list_methods(codebase_hash="abc", callee_pattern="memcpy")""",
+        timeout=30,
     )
     def list_methods(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -118,7 +119,8 @@ def list_methods(
 Examples:
     list_files(codebase_hash="abc")
     list_files(codebase_hash="abc", local_path="src/lib")
-    list_files(codebase_hash="abc", page=2)"""
+    list_files(codebase_hash="abc", page=2)""",
+        timeout=30,
     )
     def list_files(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -183,7 +185,8 @@ def list_files(
 
 Examples:
     get_method_source(codebase_hash="abc", method_name="main")
-    get_method_source(codebase_hash="abc", method_name="init", filename="driver.c")"""
+    get_method_source(codebase_hash="abc", method_name="init", filename="driver.c")""",
+        timeout=30,
     )
     def get_method_source(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -351,7 +354,8 @@ def get_method_source(
 
 Examples:
     list_calls(codebase_hash="abc", callee_pattern="strcpy")
-    list_calls(codebase_hash="abc", caller_pattern="main")"""
+    list_calls(codebase_hash="abc", caller_pattern="main")""",
+        timeout=30,
     )
     def list_calls(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -423,7 +427,8 @@ def list_calls(
 
 Examples:
     get_call_graph(codebase_hash="abc", method_name="main", direction="outgoing")
-    get_call_graph(codebase_hash="abc", method_name="vuln_func", direction="incoming")"""
+    get_call_graph(codebase_hash="abc", method_name="vuln_func", direction="incoming")""",
+        timeout=120,
     )
     def get_call_graph(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -514,7 +519,8 @@ def get_call_graph(
     - Useful for understanding function signatures.
 
 Examples:
-    list_parameters(codebase_hash="abc", method_name="login")"""
+    list_parameters(codebase_hash="abc", method_name="login")""",
+        timeout=30,
     )
     def list_parameters(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -565,7 +571,8 @@ def list_parameters(
     - Useful as a first step when exploring a new codebase.
 
 Examples:
-    get_codebase_summary(codebase_hash="abc")"""
+    get_codebase_summary(codebase_hash="abc")""",
+        timeout=30,
     )
     def get_codebase_summary(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")]
@@ -727,7 +734,8 @@ def get_codebase_summary(
     - filename should be relative to the project root (e.g., 'src/main.c' not '/absolute/path/src/main.c').
 
 Examples:
-    get_code_snippet(codebase_hash="abc", filename="main.c", start_line=10, end_line=20)"""
+    get_code_snippet(codebase_hash="abc", filename="main.c", start_line=10, end_line=20)""",
+        timeout=30,
     )
     def get_code_snippet(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -850,7 +858,8 @@ def get_code_snippet(
     - Use get_cpgql_syntax_help for reference.
 
 Examples:
-    run_cpgql_query(codebase_hash="abc", query="cpg.method.name.l")"""
+    run_cpgql_query(codebase_hash="abc", query="cpg.method.name.l")""",
+        timeout=120,
     )
     def run_cpgql_query(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -956,7 +965,8 @@ def run_cpgql_query(
     - filename in buffer_access_location should be relative to the project root (e.g., 'src/parser.c:100').
 
 Examples:
-    find_bounds_checks(codebase_hash="abc", buffer_access_location="parser.c:3393")"""
+    find_bounds_checks(codebase_hash="abc", buffer_access_location="parser.c:3393")""",
+        timeout=120,
     )
     def find_bounds_checks(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1035,7 +1045,8 @@ def find_bounds_checks(
     - Use this to learn how to write queries for run_cpgql_query.
 
 Examples:
-    get_cpgql_syntax_help()"""
+    get_cpgql_syntax_help()""",
+        timeout=30,
     )
     def get_cpgql_syntax_help() -> Dict[str, Any]:
         """Get CPGQL syntax documentation and common query patterns."""
@@ -1158,7 +1169,8 @@ def get_cpgql_syntax_help() -> Dict[str, Any]:
     - Returns plain text.
 
 Examples:
-    get_cfg(codebase_hash="abc", method_name="main")"""
+    get_cfg(codebase_hash="abc", method_name="main")""",
+        timeout=120,
     )
     def get_cfg(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1237,7 +1249,8 @@ def get_cfg(
     - Does not read header files; uses CPG type info.
 
 Examples:
-    get_type_definition(codebase_hash="abc", type_name=".*request_t.*")"""
+    get_type_definition(codebase_hash="abc", type_name=".*request_t.*")""",
+        timeout=30,
     )
     def get_type_definition(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1331,7 +1344,8 @@ def get_type_definition(
     - filename should be relative to the project root (e.g., 'src/main.c').
 
 Examples:
-    get_macro_expansion(codebase_hash="abc", filename="main.c", line_number=42)"""
+    get_macro_expansion(codebase_hash="abc", filename="main.c", line_number=42)""",
+        timeout=30,
     )
     def get_macro_expansion(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1483,7 +1497,8 @@ def get_macro_expansion(
 
 Examples:
     discover_fixed_vulnerabilities(codebase_hash="abc")
-    discover_fixed_vulnerabilities(codebase_hash="abc", limit=100)"""
+    discover_fixed_vulnerabilities(codebase_hash="abc", limit=100)""",
+        timeout=120,
     )
     def discover_fixed_vulnerabilities(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],

src/tools/core_tools.py

@@ -386,7 +386,8 @@ def register_core_tools(mcp, services: dict):
         source_type="github",
         source_path="https://github.com/joernio/sample-repo",
         language="java"
-    )"""
+    )""",
+        timeout=600,
     )
     async def generate_cpg(
         source_type: Annotated[str, Field(description="Either 'local' or 'github'")],
@@ -629,7 +630,8 @@ async def generate_cpg(
     - If status is 'generating', wait and retry.
 
 Examples:
-    get_cpg_status(codebase_hash="abc123456789")"""
+    get_cpg_status(codebase_hash="abc123456789")""",
+        timeout=30,
     )
     def get_cpg_status(
         codebase_hash: Annotated[str, Field(description="The hash identifier of the codebase")]

src/tools/export_tools.py

@@ -758,7 +758,8 @@ def register_export_tools(mcp, services: dict):
     - Parses text outputs into structured data
     - Assigns HIGH confidence to confirmed vulnerabilities
     - Maps findings to CWE IDs
-    - Determines severity based on vulnerability type"""
+    - Determines severity based on vulnerability type""",
+        timeout=60,
     )
     def store_findings(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -919,7 +920,8 @@ def store_findings(
     - Only exports findings matching severity/confidence filters
     - Includes code flow paths for taint flows
     - Maps findings to CWE IDs
-    - Compatible with GitHub Code Scanning upload"""
+    - Compatible with GitHub Code Scanning upload""",
+        timeout=60,
     )
     def export_sarif(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],

src/tools/taint_analysis_tools.py

@@ -422,7 +422,8 @@ def register_taint_analysis_tools(mcp, services: dict):
 
 Examples:
     find_taint_sources(codebase_hash="abc", language="c")
-    find_taint_sources(codebase_hash="abc", source_patterns=["read_from_socket"])"""
+    find_taint_sources(codebase_hash="abc", source_patterns=["read_from_socket"])""",
+        timeout=120,
     )
     def find_taint_sources(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -548,7 +549,8 @@ def _execute():
 
 Examples:
     find_taint_sinks(codebase_hash="abc", language="c")
-    find_taint_sinks(codebase_hash="abc", sink_patterns=["custom_exec"])"""
+    find_taint_sinks(codebase_hash="abc", sink_patterns=["custom_exec"])""",
+        timeout=120,
     )
     def find_taint_sinks(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -697,7 +699,8 @@ def _execute():
 
     # Manual mode — specific source and sink
     find_taint_flows(codebase_hash="...", source_location="main.c:42", sink_location="utils.c:100")
-    find_taint_flows(codebase_hash="...", source_node_id=12345, sink_node_id=67890)"""
+    find_taint_flows(codebase_hash="...", source_node_id=12345, sink_node_id=67890)""",
+        timeout=300,
     )
     def find_taint_flows(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -932,7 +935,8 @@ def _execute():
 Examples:
     get_program_slice(codebase_hash="abc", location="main.c:42")
     get_program_slice(codebase_hash="abc", location="parser.c:500:memcpy", direction="backward", max_depth=3)
-    get_program_slice(codebase_hash="abc", location="module/file.c:100", direction="forward")"""
+    get_program_slice(codebase_hash="abc", location="module/file.c:100", direction="forward")""",
+        timeout=300,
     )
     def get_program_slice(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1046,7 +1050,8 @@ def _execute():
     - location filename should be relative to the project root.
 
 Examples:
-    get_variable_flow(codebase_hash="abc", location="main.c:50", variable="len", direction="backward")"""
+    get_variable_flow(codebase_hash="abc", location="main.c:50", variable="len", direction="backward")""",
+        timeout=120,
     )
     def get_variable_flow(
         codebase_hash: str,
@@ -1159,7 +1164,8 @@ def _execute():
 Notes:
     - Deep interprocedural analysis can be slow (~2 min for large codebases).
     - Use get_program_slice to understand control flow around specific locations.
-    - Use find_taint_flows for alternative dataflow analysis approach."""
+    - Use find_taint_flows for alternative dataflow analysis approach.""",
+        timeout=300,
     )
     def find_use_after_free(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1244,7 +1250,8 @@ def _execute():
     Human-readable text showing:
     - Each potential double-free issue with pointer name
     - First and second free locations with [file:line]
-    - Flow type (same-ptr, alias, or [CROSS-FUNC])"""
+    - Flow type (same-ptr, alias, or [CROSS-FUNC])""",
+        timeout=300,
     )
     def find_double_free(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1340,7 +1347,8 @@ def _execute():
 Notes:
     - Includes deep interprocedural analysis using Joern's dataflow engine.
     - Use get_program_slice for deeper control-flow context around specific locations.
-    - Use find_taint_flows to check if allocation arguments come from external input."""
+    - Use find_taint_flows to check if allocation arguments come from external input.""",
+        timeout=300,
     )
     def find_null_pointer_deref(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],
@@ -1438,7 +1446,8 @@ def _execute():
 Notes:
     - Includes deep interprocedural analysis using Joern's dataflow engine.
     - Use get_program_slice for deeper control-flow context around specific locations.
-    - Use find_taint_flows to check if arithmetic operands come from external input."""
+    - Use find_taint_flows to check if arithmetic operands come from external input.""",
+        timeout=300,
     )
     def find_integer_overflow(
         codebase_hash: Annotated[str, Field(description="The codebase hash from generate_cpg")],