LangChain Agent Shared Wiring

kind get kubeconfig --name finops-cluster --internal > /tmp/kubeconfig-docker.yaml

docker run --rm -p 8000:8000 --network kind \
  -v /tmp/kubeconfig-docker.yaml:/home/appuser/.kube/config:ro \
  -e ENABLE_UNSAFE_STREAMABLE_HTTP_TRANSPORT=1 \
  -e PORT=8000 \
  <other-docker-args>

"""Shared LangChain + MCP helpers for example scripts.

Expects a persistent HTTP MCP container (native Streamable HTTP on
http://localhost:8000/mcp). No Supergateway required.
"""

import json
import os
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from langchain_mcp_tools import convert_mcp_to_langchain_tools

# Load the .env file located at the project root (reached here via Path(__file__).parents[3])
load_dotenv(Path(__file__).parents[3] / ".env")

MCP_URL = os.getenv("K8S_MCP_URL", "http://localhost:8000/mcp")
MCP_SERVERS = {
    "k8s": {
        "url": MCP_URL,
        "transport": "streamable_http",
    }
}


def decode_tool_result(result: Any) -> dict:
    """Normalize MCP tool output to a dictionary.

    MCP tools may return a dict already or a JSON string. This helper
    converts either shape into a dict with a predictable structure.
    """
    if isinstance(result, dict):
        return result
    text = str(result).strip()
    if not text:
        return {"items": []}
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"items": []}


def tool_by_name(tools: list[BaseTool], name: str) -> BaseTool:
    """Find a LangChain tool by name in the list returned from MCP."""
    tool = next((t for t in tools if t.name == name), None)
    if tool is None:
        raise RuntimeError(f"MCP tool not found: {name}")
    return tool


async def get_mcp_tools():
    """Convert MCP-discovered tools into LangChain tools.

    Returns a tuple (tools, cleanup) where `cleanup` should be awaited
    when tools are no longer needed.
    """
    return await convert_mcp_to_langchain_tools(MCP_SERVERS)


def build_llm(max_tokens: int | None = None) -> ChatOpenAI:
    """Construct the ChatOpenAI LLM from environment variables."""
    return ChatOpenAI(
        model=os.environ["OPENAI_MODEL_ID"],
        base_url=os.environ["OPENAI_BASE_URL"],
        api_key=os.environ["OPENAI_API_KEY"],
        temperature=float(os.getenv("OPENAI_TEMPERATURE", "0.3")),
        max_tokens=max_tokens or int(os.getenv("OPENAI_MAX_TOKENS", "512")),
    )


async def run_agent(prompt: str, max_tokens: int | None = None) -> str:
    """Full agent loop: discover tools, create the agent, invoke it, return final message.

    This function:
    - Converts MCP tools to LangChain tools
    - Builds the LLM using environment variables
    - Creates a LangChain agent with the tools
    - Invokes the agent with a single HumanMessage containing the prompt
    - Returns the last assistant message content
    - Ensures cleanup is always run
    """
    tools, cleanup = await get_mcp_tools()
    try:
        agent = create_agent(build_llm(max_tokens=max_tokens), tools)
        result = await agent.ainvoke({"messages": [HumanMessage(content=prompt)]})
        return result["messages"][-1].content
    finally:
        await cleanup()

import asyncio
from mcp_client import run_agent

async def main():
    prompt = "List all running pods in the finops namespace and summarize any errors."
    output = await run_agent(prompt, max_tokens=256)
    print(output)

asyncio.run(main())