Running agent frameworks on kagent: ADK, LangGraph, CrewAI and AutoGen, behind agentgateway

apiVersion: kagent.dev/v1alpha2
kind: Agent
metadata: { name: sre-crew-kagent, namespace: kagent }
spec:
  type: Declarative
  declarative:
    modelConfig: default-model-config
    systemMessage: |
      Resolve the incident in three steps: delegate to sre-diagnostician for the
      root cause, then sre-remediation-planner for the exact image patch, then
      apply it with patch_deployment_image after the user approves.
    tools:
      - type: Agent
        agent: { name: sre-diagnostician }
      - type: Agent
        agent: { name: sre-remediation-planner }
      - type: McpServer
        mcpServer:
          apiGroup: kagent.dev
          kind: RemoteMCPServer
          name: k8s-ops
          toolNames: [ patch_deployment_image ]
          requireApproval: [ patch_deployment_image ]   # the Reviewer role

from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.models.lite_llm import LiteLlm
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StreamableHTTPConnectionParams

def model():   # LiteLlm routes openai/ to the gateway's OpenAI-compatible endpoint
    return LiteLlm(model=f"openai/{MODEL}", api_base=LLM_BASE_URL, api_key="sk-gateway")
def tools():
    return MCPToolset(connection_params=StreamableHTTPConnectionParams(url=MCP_URL))

diagnostician = LlmAgent(
    name="diagnostician", model=model(), tools=[tools()],
    description="Finds the root cause of a failing workload from cluster state.",
    instruction=(
        "Inspect the failing workload in the incident namespace with your tools "
        "(pods, events, logs, deployment spec) and state the single root cause. "
        "You diagnose only."),
)
planner = LlmAgent(
    name="planner", model=model(), tools=[tools()],
    description="Turns a root cause into one exact image patch.",
    instruction=(
        "Given the diagnosis, state the exact remediation: the namespace, deployment "
        "name, container name, and a valid image tag to set. Use describe_deployment "
        "to confirm the current container first. Do not apply it."),
)
operator = LlmAgent(
    name="operator", model=model(), tools=[tools()],
    description="Applies the agreed image patch.",
    instruction=(
        "Apply the planned fix by calling patch_deployment_image with the agreed "
        "namespace, deployment, container and image, then confirm what changed."),
)

# SequentialAgent runs the three in order, sharing session state, so the operator
# reliably runs after the plan. (An LlmAgent coordinator with sub_agents can hand
# off and stop after the plan; the workflow agent guarantees the order.)
root_agent = SequentialAgent(
    name="sre_coordinator",
    sub_agents=[diagnostician, planner, operator],
)

apiVersion: kagent.dev/v1alpha2
kind: Agent
metadata: { name: sre-crew-adk, namespace: kagent }
spec:
  type: BYO
  byo:
    deployment:
      image: sre-crew-adk:dev
      env:
        - { name: LLM_BASE_URL, value: "http://frameworks-gw.agentgateway-system.svc.cluster.local/v1" }
        - { name: MCP_URL,      value: "http://frameworks-gw.agentgateway-system.svc.cluster.local/mcp" }
        - { name: MODEL,        value: "claude-haiku-4-5" }

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

# The instructions are three system prompts, one per role. Each node prepends its
# own prompt to the running conversation before calling the model.
DIAGNOSE_SYS = (
    "You are a Kubernetes diagnostician for the incident namespace. Inspect the "
    "failing workload with your tools (pods, events, logs, deployment spec) and "
    "state the single root cause. When confident, stop calling tools and reply "
    "with the root cause in one or two sentences.")
PLAN_SYS = (
    "You are an SRE remediation planner. From the diagnosis above, call "
    "patch_deployment_image with the exact namespace, deployment, container and a "
    "valid image tag that fixes the incident.")
SUMMARIZE_SYS = (
    "Write a short incident summary: what broke, the root cause, and the fix that "
    "was applied (or that the reviewer rejected). Three sentences at most.")

# Three model clients, all reaching Claude through the gateway. The diagnostician
# gets the read tools; the planner is forced to emit exactly one patch proposal.
llm_diagnose  = ChatOpenAI(model=MODEL, base_url=LLM_BASE_URL, api_key="sk-gateway").bind_tools(READ_TOOLS)
llm_plan      = ChatOpenAI(model=MODEL, base_url=LLM_BASE_URL, api_key="sk-gateway").bind_tools(
                    [patch_tool], tool_choice="patch_deployment_image")
llm_summarize = ChatOpenAI(model=MODEL, base_url=LLM_BASE_URL, api_key="sk-gateway")

async def diagnose(state):            # Diagnostician: prepend DIAGNOSE_SYS, reason, call read tools
    return {"messages": [await llm_diagnose.ainvoke([HumanMessage(DIAGNOSE_SYS)] + state["messages"])]}

async def run_read_tools(state):      # execute the read tools the diagnostician asked for
    return {"messages": await call_tools(state["messages"][-1].tool_calls)}

def after_diagnose(state):            # loop until no more tool calls, then move to plan
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else "plan"

async def plan(state):                # Planner: PLAN_SYS forces one patch_deployment_image call
    return {"messages": [await llm_plan.ainvoke([HumanMessage(PLAN_SYS)] + state["messages"])]}

async def review(state):              # Reviewer: pause for human approval, then apply or reject
    call = state["messages"][-1].tool_calls[0]            # the proposed patch
    decision = interrupt({"action_requests": [
        {"name": call["name"], "args": call["args"], "id": call["id"]}]})
    return apply_patch(call) if decision.get("decision_type") == "approve" else reject(call)

async def summarize(state):           # write the incident summary using SUMMARIZE_SYS
    return {"messages": [await llm_summarize.ainvoke([HumanMessage(SUMMARIZE_SYS)] + state["messages"])]}

g = StateGraph(CrewState)
g.add_node("diagnose", diagnose)      # find the root cause (loops through the read tools)
g.add_node("tools", run_read_tools)
g.add_node("plan", plan)              # propose the exact image patch
g.add_node("review", review)          # human approves, then the patch is applied
g.add_node("summarize", summarize)
g.add_edge(START, "diagnose")
g.add_conditional_edges("diagnose", after_diagnose, {"tools": "tools", "plan": "plan"})
g.add_edge("tools", "diagnose")       # back to the diagnostician after each tool round
g.add_edge("plan", "review")
g.add_edge("review", "summarize")
g.add_edge("summarize", END)
graph = g.compile(checkpointer=MemorySaver())

from crewai import LLM, Agent, Crew, Process, Task
from crewai_tools import MCPServerAdapter

llm   = LLM(model=f"openai/{MODEL}", base_url=LLM_BASE_URL, api_key="sk-gateway")
tools = MCPServerAdapter({"url": MCP_URL, "transport": "streamable-http"}).tools

diagnostician = Agent(role="Kubernetes Diagnostician", llm=llm, tools=tools,
    goal="Find the single root cause of the failing workload in the incident namespace.",
    backstory="An SRE who reads pod state, events and logs to pinpoint why a workload will not start.")
planner = Agent(role="Remediation Planner", llm=llm, tools=tools,
    goal="Turn the root cause into one concrete, minimal image patch.",
    backstory="An SRE who proposes the smallest safe change: the deployment, container and exact image tag.")
operator = Agent(role="SRE Operator", llm=llm, tools=tools,
    goal="Apply the agreed image patch so the workload recovers.",
    backstory="An operator who executes the approved remediation against the cluster.")

crew = Crew(agents=[diagnostician, planner, operator],
            tasks=[diagnose, plan, apply], process=Process.sequential)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StreamableHttpServerParams, mcp_server_tools

client = OpenAIChatCompletionClient(model=MODEL, base_url=LLM_BASE_URL, api_key="sk-gateway",
                                    model_info=ModelInfo(function_calling=True, ...))
tools  = await mcp_server_tools(StreamableHttpServerParams(url=MCP_URL))

team = RoundRobinGroupChat([
    AssistantAgent("diagnostician", client, tools=tools, system_message=(
        "Inspect the failing workload in the incident namespace (pods, events, logs, "
        "deployment spec) and state the single root cause. Diagnose only.")),
    AssistantAgent("planner", client, tools=tools, system_message=(
        "From the diagnosis, give the exact image patch: namespace, deployment, "
        "container and a valid image tag. Use describe_deployment to confirm. Do not apply it.")),
    AssistantAgent("operator", client, tools=tools, system_message=(
        "Apply the planned fix with patch_deployment_image, then write a one-line "
        "summary and end your message with TERMINATE.")),
], termination_condition=TextMentionTermination("TERMINATE"))

apiVersion: enterpriseagentgateway.solo.io/v1alpha1
kind: EnterpriseAgentgatewayPolicy
metadata: { name: llm-prompt-guard, namespace: agentgateway-system }
spec:
  targetRefs:
    - { group: gateway.networking.k8s.io, kind: HTTPRoute, name: llm-route }
  backend:
    ai:
      promptGuard:
        request:
          - response: { statusCode: 403, message: "Blocked by prompt guard" }
            regex:
              action: Reject
              matches:
                - "(?i)ignore (all )?(your )?(previous|prior|earlier|above) instructions"
                - "(?i)(reveal|show|print|repeat) (your |the )?system prompt"

"""k8s-ops: MCP server over the live Kubernetes API, RBAC-scoped to `incident`.
Served at /mcp (FastMCP streamable-http), reached only through agentgateway."""
import contextlib, os
from kubernetes import client, config
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route

# We sit behind agentgateway, so disable FastMCP's DNS-rebinding guard (it would
# reject the in-cluster gateway Host header with a 421).
_TS = TransportSecuritySettings(enable_dns_rebinding_protection=False)
try:
    config.load_incluster_config()
except config.ConfigException:
    config.load_kube_config()
_core, _apps = client.CoreV1Api(), client.AppsV1Api()
mcp = FastMCP("k8s-ops", stateless_http=True, transport_security=_TS)


@mcp.tool()
def get_pods(namespace: str = "incident") -> dict:
    """List pods with phase, restarts, and why a container is not ready
    (e.g. ImagePullBackOff or CrashLoopBackOff)."""
    out = []
    for p in _core.list_namespaced_pod(namespace).items:
        reasons, restarts = [], 0
        for cs in p.status.container_statuses or []:
            restarts += cs.restart_count or 0
            st = cs.state
            if st and st.waiting and st.waiting.reason:    reasons.append(st.waiting.reason)
            if st and st.terminated and st.terminated.reason: reasons.append(st.terminated.reason)
        out.append({"name": p.metadata.name, "phase": p.status.phase,
                    "restarts": restarts, "reasons": reasons})
    return {"namespace": namespace, "pods": out}


@mcp.tool()
def get_events(namespace: str = "incident") -> dict:
    """Recent events: image pull errors, failed scheduling, OOMKills, etc."""
    items = [{"type": e.type, "reason": e.reason,
              "object": f"{e.involved_object.kind}/{e.involved_object.name}",
              "message": e.message, "count": e.count}
             for e in _core.list_namespaced_event(namespace).items]
    return {"namespace": namespace, "events": items[-40:]}


@mcp.tool()
def get_pod_logs(namespace: str, pod: str, tail_lines: int = 50) -> dict:
    """Tail a pod's logs. No logs (container never started) is itself a signal."""
    try:
        logs = _core.read_namespaced_pod_log(name=pod, namespace=namespace, tail_lines=tail_lines)
    except client.ApiException as e:
        logs = f"(no logs available: {e.reason})"
    return {"namespace": namespace, "pod": pod, "logs": logs}


@mcp.tool()
def describe_deployment(namespace: str, name: str) -> dict:
    """Describe a Deployment: container images, replica readiness, conditions."""
    d = _apps.read_namespaced_deployment(name=name, namespace=namespace)
    return {"namespace": namespace, "name": name,
            "containers": [{"name": c.name, "image": c.image} for c in d.spec.template.spec.containers],
            "replicas": {"desired": d.spec.replicas, "ready": d.status.ready_replicas or 0,
                         "available": d.status.available_replicas or 0}}


@mcp.tool()
def patch_deployment_image(namespace: str, name: str, container: str, image: str) -> dict:
    """Set a container's image on a Deployment. The one mutating tool: the fix.
    Behind the gateway this is the call an ext-auth HITL policy can park for review."""
    body = {"spec": {"template": {"spec": {"containers": [{"name": container, "image": image}]}}}}
    _apps.patch_namespaced_deployment(name=name, namespace=namespace, body=body)
    return {"patched": f"{namespace}/{name}", "container": container, "image": image}


async def health(_req):
    return JSONResponse({"status": "ok"})

# FastMCP's session manager must be active before requests; mounted at "/" so the
# MCP endpoint lands at /mcp, with /healthz alongside for the readiness probe.
@contextlib.asynccontextmanager
async def lifespan(_app):
    async with contextlib.AsyncExitStack() as stack:
        await stack.enter_async_context(mcp.session_manager.run())
        yield

app = Starlette(routes=[Route("/healthz", health),
                        Mount("/", app=mcp.streamable_http_app())], lifespan=lifespan)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))

apiVersion: kagent.dev/v1alpha2
kind: RemoteMCPServer
metadata: { name: k8s-ops, namespace: kagent }
spec:
  description: "Kubernetes ops tools for the incident namespace, fronted by agentgateway."
  protocol: STREAMABLE_HTTP
  # the gateway /mcp route, never the pod directly
  url: http://frameworks-gw.agentgateway-system.svc.cluster.local/mcp
  timeout: 10m0s
  sseReadTimeout: 10m0s
  terminateOnClose: true

apiVersion: v1
kind: ServiceAccount
metadata: { name: k8s-ops, namespace: incident }
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata: { name: k8s-ops, namespace: incident }
rules:
  - apiGroups: [""]
    resources: [pods, pods/log, events]
    verbs: [get, list, watch]
  - apiGroups: [apps]
    resources: [deployments]
    verbs: [get, list, watch, patch]   # patch = the one mutating verb
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata: { name: k8s-ops, namespace: incident }
roleRef: { apiGroup: rbac.authorization.k8s.io, kind: Role, name: k8s-ops }
subjects:
  - { kind: ServiceAccount, name: k8s-ops, namespace: incident }
---
apiVersion: apps/v1
kind: Deployment
metadata: { name: k8s-ops, namespace: incident, labels: { app: k8s-ops } }
spec:
  replicas: 1
  selector: { matchLabels: { app: k8s-ops } }
  template:
    metadata: { labels: { app: k8s-ops } }
    spec:
      serviceAccountName: k8s-ops
      containers:
        - name: k8s-ops
          image: k8s-ops-mcp:dev
          imagePullPolicy: IfNotPresent
          ports: [ { containerPort: 8080, name: http } ]
          readinessProbe: { httpGet: { path: /healthz, port: 8080 }, periodSeconds: 3 }
---
apiVersion: v1
kind: Service
metadata: { name: k8s-ops, namespace: incident, labels: { app: k8s-ops } }
spec:
  selector: { app: k8s-ops }
  ports: [ { name: http, port: 8080, targetPort: 8080, appProtocol: http } ]