agent-factory: generate agent auto-deploy-application-to-developm

app/agent.py

@@ -0,0 +1,148 @@
+"""
+AutoDeployApplicationToDevelopm Agent — auto-generated by Agent Factory.
+"""
+import asyncio
+import logging
+import os
+
+import click
+import httpx
+import uvicorn
+from contextlib import asynccontextmanager
+from dotenv import load_dotenv
+from langchain_core.rate_limiters import InMemoryRateLimiter
+from langchain_openai import AzureChatOpenAI
+
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+from app.config import (
+    AGENT_SELF_URL,
+    AZURE_OPENAI_API_KEY,
+    AZURE_OPENAI_API_VERSION,
+    AZURE_OPENAI_DEPLOYMENT,
+    AZURE_OPENAI_ENDPOINT,
+    LOG_LEVEL,
+    REGISTRY_URL,
+)
+from app.skills import AGENT_CONFIG, AUTO_DEPLOY_APPLICATION_TO_DEVELOPM_SKILLS
+from app.workflows.auto_deploy_application_to_developm_workflow import create_auto_deploy_application_to_developm_workflow
+
+load_dotenv()
+
+# ── Logging ──────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=getattr(logging, LOG_LEVEL.upper(), logging.INFO),
+    format="%(asctime)s %(name)s %(levelname)s %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# ── LLM ──────────────────────────────────────────────────────────────────
+rate_limiter = InMemoryRateLimiter(
+    requests_per_second=10 / 60,
+    check_every_n_seconds=0.1,
+    max_bucket_size=10,
+)
+
+llm = AzureChatOpenAI(
+    temperature=0,
+    azure_deployment=AZURE_OPENAI_DEPLOYMENT,
+    api_version=AZURE_OPENAI_API_VERSION,
+    azure_endpoint=AZURE_OPENAI_ENDPOINT or "",
+    api_key=AZURE_OPENAI_API_KEY or "",
+    max_retries=5,
+    timeout=120,
+    rate_limiter=rate_limiter,
+)
+
+workflow = create_auto_deploy_application_to_developm_workflow(llm)
+
+
+# ── Endpoints ────────────────────────────────────────────────────────────
+
+async def health_check(request: Request) -> JSONResponse:
+    return JSONResponse({"status": "healthy", "agent": 'AutoDeployApplicationToDevelopm'})
+
+
+async def agent_manifest(request: Request) -> JSONResponse:
+    """GET /.well-known/agent.json"""
+    return JSONResponse({
+        "name": AGENT_CONFIG["name"],
+        "version": AGENT_CONFIG["version"],
+        "description": AGENT_CONFIG["description"],
+        "url": "/",
+        "skills": [
+            {
+                "id": s.id,
+                "name": s.name,
+                "description": s.description,
+                "tags": s.tags,
+                "inputSchema": AGENT_CONFIG.get("input_schema"),
+                "outputSchema": AGENT_CONFIG.get("output_schema"),
+            } for s in AUTO_DEPLOY_APPLICATION_TO_DEVELOPM_SKILLS
+        ],
+        "capabilities": AGENT_CONFIG["capabilities"],
+    })
+
+
+async def process_endpoint(request: Request) -> JSONResponse:
+    """POST /process — run the agent workflow."""
+    try:
+        body = await request.json()
+        result = await workflow.ainvoke(body)
+        return JSONResponse(result)
+    except Exception as exc:
+        logger.error("Processing failed: %s", exc, exc_info=True)
+        return JSONResponse({"error": str(exc)}, status_code=500)
+
+
+# ── Self-registration ────────────────────────────────────────────────────
+
+async def _register_with_registry():
+    if not REGISTRY_URL:
+        logger.info("REGISTRY_URL not set — skipping self-registration")
+        return
+    await asyncio.sleep(2)
+    url = f"{REGISTRY_URL.rstrip('/')}/agents/register-url"
+    for attempt in range(3):
+        try:
+            async with httpx.AsyncClient(timeout=10.0) as client:
+                resp = await client.post(url, json={"endpoint": AGENT_SELF_URL})
+                if resp.status_code in (200, 201):
+                    logger.info("Self-registered with agent-registry at %s", REGISTRY_URL)
+                    return
+                logger.warning("Registration attempt %d: HTTP %d", attempt + 1, resp.status_code)
+        except Exception as exc:
+            logger.warning("Registration attempt %d failed: %s", attempt + 1, exc)
+        await asyncio.sleep(5)
+    logger.error("Failed to self-register after 3 attempts")
+
+
+@asynccontextmanager
+async def lifespan(app):
+    task = asyncio.create_task(_register_with_registry())
+    yield
+    task.cancel()
+
+
+app = Starlette(
+    routes=[
+        Route("/health", methods=["GET"], endpoint=health_check),
+        Route("/.well-known/agent.json", methods=["GET"], endpoint=agent_manifest),
+        Route("/process", methods=["POST"], endpoint=process_endpoint),
+    ],
+    lifespan=lifespan,
+)
+
+
+@click.command()
+@click.option("--host", default="0.0.0.0")
+@click.option("--port", default=8080, type=int)
+def main(host: str, port: int):
+    uvicorn.run(app, host=host, port=port, log_level=LOG_LEVEL.lower())
+
+
+if __name__ == "__main__":
+    main()

app/config.py

@@ -0,0 +1,21 @@
+"""
+Configuration for the AutoDeployApplicationToDevelopm agent.
+"""
+import os
+
+# ── Azure OpenAI ─────────────────────────────────────────────────────────
+AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
+AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
+AZURE_OPENAI_API_VERSION = os.getenv("AZURE_OPENAI_API_VERSION", "2024-08-01-preview")
+AZURE_OPENAI_DEPLOYMENT = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o")
+
+# ── Agent Registry ───────────────────────────────────────────────────────
+REGISTRY_URL = os.getenv(
+    "REGISTRY_URL", "http://agent-gateway.agents.svc.cluster.local"
+)
+AGENT_SELF_URL = os.getenv(
+    "AGENT_SELF_URL", "http://auto-deploy-application-to-developm.agents.svc.cluster.local"
+)
+
+# ── Logging ──────────────────────────────────────────────────────────────
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")

app/nodes/core_node.py

@@ -0,0 +1,43 @@
+async def process(state: dict) -> dict:
+    """
+    Deploys an application to the development environment using the provided CI pipeline configuration
+    and PostgreSQL version. Utilizes an LLM to generate deployment instructions and returns the deployment
+    status and URL.
+
+    Args:
+        state (dict): A dictionary containing the input fields:
+            - ci_pipeline_config_path (str): Path to the CI pipeline configuration file.
+            - postgres_version (str): Version of PostgreSQL to use.
+
+    Returns:
+        dict: A dictionary containing the output fields:
+            - deployment_status (str): Status of the deployment (e.g., "success", "failure").
+            - deployment_url (str): URL of the deployed application.
+    """
+    from app.agent import llm
+    from langchain_core.messages import SystemMessage, HumanMessage
+
+    try:
+        ci_pipeline_config_path = state.get("ci_pipeline_config_path", "")
+        postgres_version = state.get("postgres_version", "")
+
+        if not ci_pipeline_config_path or not postgres_version:
+            raise ValueError("Both 'ci_pipeline_config_path' and 'postgres_version' must be provided.")
+
+        messages = [
+            SystemMessage(content="You are a deployment assistant. Your task is to deploy applications to a development environment."),
+            HumanMessage(content=f"Deploy the application using the CI pipeline configuration at '{ci_pipeline_config_path}' and PostgreSQL version '{postgres_version}'. Provide the deployment status and URL.")
+        ]
+
+        response = await llm.ainvoke(messages)
+        deployment_details = response.content.strip().split("\n")
+        
+        if len(deployment_details) < 2:
+            raise ValueError("Unexpected response format from LLM.")
+
+        deployment_status = deployment_details[0].strip()
+        deployment_url = deployment_details[1].strip()
+
+        return {"deployment_status": deployment_status, "deployment_url": deployment_url, "phase": "complete"}
+    except Exception as exc:
+        return {"error": str(exc), "phase": "failed"}

app/skills.py

@@ -0,0 +1,29 @@
+"""
+A2A skill declarations for AutoDeployApplicationToDevelopm.
+"""
+from a2a.types import AgentSkill
+
+
+AUTO_DEPLOY_APPLICATION_TO_DEVELOPM_SKILLS = [
+    AgentSkill(
+        id="auto_deploy_application_to_developm_skill",
+        name="AutoDeployApplicationToDevelopm",
+        description="Deploy application to development environment",
+        tags=["auto-generated"],
+        examples=[],
+    ),
+]
+
+
+AGENT_CONFIG = {
+    "name": "AutoDeployApplicationToDevelopm",
+    "description": "Deploy claims-api to development environment with Postgres database.",
+    "version": "1.0.0",
+    "framework": "LangGraph + Starlette",
+    "capabilities": {
+        "streaming": False,
+        "async": True,
+    },
+    "input_schema": {"ci_pipeline_config_path": "string", "postgres_version": "string"},
+    "output_schema": {"deployment_status": "string", "deployment_url": "string"},
+}

app/states/auto_deploy_application_to_developm_state.py

@@ -0,0 +1,17 @@
+"""
+State definitions for AutoDeployApplicationToDevelopm agent.
+"""
+from __future__ import annotations
+from typing import Any, Dict, List, Optional
+from langgraph.graph import MessagesState
+
+
+class AutoDeployApplicationToDevelopmState(MessagesState):
+    """Workflow state for AutoDeployApplicationToDevelopm."""
+    # ── Input fields ─────────────────────────────────────────────────────
+    pass
+    # ── Output fields ────────────────────────────────────────────────────
+    pass
+    # ── Internal ─────────────────────────────────────────────────────────
+    error: Optional[str]
+    phase: str

app/workflows/auto_deploy_application_to_developm_workflow.py

@@ -0,0 +1,19 @@
+"""
+LangGraph workflow for AutoDeployApplicationToDevelopm agent.
+"""
+from langgraph.graph import StateGraph, END
+
+from app.states.auto_deploy_application_to_developm_state import AutoDeployApplicationToDevelopmState
+from app.nodes.core_node import process
+
+
+def create_auto_deploy_application_to_developm_workflow(llm):
+    """Build and compile the AutoDeployApplicationToDevelopm workflow graph."""
+    graph = StateGraph(AutoDeployApplicationToDevelopmState)
+
+    graph.add_node("process", process)
+
+    graph.set_entry_point("process")
+    graph.add_edge("process", END)
+
+    return graph.compile()