Upload tools_py.py

tools_py.py

@@ -0,0 +1,83 @@
+# -*- coding: utf-8 -*-
+"""tools.py
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1Srm2oS1q5iYwlo4PoMx6TBc2uNoSVao3
+"""
+
+from typing import Dict, Any, List
+from langchain.tools import Tool
+
+"""**It's purely deterministic Python, no LLM inside the tool, so the reasoning stays transparent.**"""
+
+def modernize_stack(legacy_stack: Dict[str, Any]) -> List[str]:
+    """Return concrete modernization actions based on a given legacy stack."""
+    actions: List[str] = []
+
+    scheduler = str(legacy_stack.get("scheduler", "")).lower()
+    language = str(legacy_stack.get("language", "")).lower()
+    architecture = str(legacy_stack.get("architecture", "")).lower()
+    dependencies = [d.lower() for d in legacy_stack.get("dependencies", [])]
+
+    if "autosys" in scheduler:
+        actions.append("Replace Autosys with Apache Airflow or AWS Step Functions.")
+
+    if "java 8" in language:
+        actions.append("Upgrade Java 8 → Java 17 and containerize the application.")
+
+    if "monolith" in architecture:
+        actions.append("Refactor monolith into microservices and add an API Gateway.")
+
+    if any("ftp" in dep for dep in dependencies):
+        actions.append("Replace FTP with S3 and event-driven ingestion via Lambda.")
+
+    actions.append("Adopt Terraform for Infrastructure-as-Code.")
+
+    return actions
+
+"""**This tool takes both the legacy_stack and an outage_scenario string to generate resilience test ideas.**"""
+
+def generate_resilience_plan(legacy_stack: Dict[str, Any], outage_scenario: str) -> List[str]:
+    """Return a resilience test plan for a given outage scenario."""
+    plan: List[str] = []
+
+    if "oracle" in outage_scenario.lower():
+        plan.append("Test RDS Multi-AZ failover for Oracle DB.")
+        plan.append("Implement circuit breaker around DB calls.")
+
+    if "kafka" in outage_scenario.lower():
+        plan.append("Simulate Kafka broker outage and validate consumer recovery.")
+
+    plan.append("Perform chaos engineering tests for network partition.")
+    plan.append("Run DR simulation and validate RTO/RPO.")
+
+    return plan
+
+"""**This step is critical, it's how the agent knows these Python functions exist and what they do.**
+
+- Deterministic: No AI hallucinations in the tools, pure logic.
+
+- Composable: We can add more tools later (cost analysis, migration path scorer).
+
+- Agent-Controlled: The agent can decide when to call each tool based on the request.
+"""
+
+def get_tools() -> List[Tool]:
+    """Return a list of LangChain Tools for the agent."""
+    return [
+        Tool(
+            name="modernize_stack",
+            func=lambda x: "\n".join(modernize_stack(x)),
+            description="Recommend modernization actions for a given legacy stack."
+        ),
+        Tool(
+            name="generate_resilience_plan",
+            func=lambda x: "\n".join(generate_resilience_plan(
+                x.get("legacy_stack", {}),
+                x.get("outage_scenario", "")
+            )),
+            description="Generate a resilience test plan for the given legacy stack and outage scenario."
+        )
+    ]