Sample: LangGraph Agent — fitness-tracker

This page shows the LangGraph code that agentspec generate produces for the fitness-tracker agent manifest. The manifest is the source of truth; this code is generated, not hand-maintained.

Try it yourself

agentspec generate examples/fitness-tracker/agent.yaml --framework langgraph

---

agent.yaml

apiVersion: agentspec.io/v1
kind: AgentSpec
metadata:
  name: fitness-tracker
  version: "1.2.0"
  description: "LangGraph fitness coaching agent with full SDK integration"
spec:
  model:
    provider: groq
    id: llama-3.3-70b-versatile
    apiKey: $env:GROQ_API_KEY
  prompts:
    system: "You are an expert fitness coach. Create personalised workout plans, track sessions, and monitor progress."
  tools:
    - name: plan-workout
      type: function
      description: "Generate a personalised workout plan based on user goals and fitness level"
    - name: log-session
      type: function
      description: "Record a completed workout session with exercises, sets, reps, and weight"
    - name: get-progress
      type: function
      description: "Retrieve user progress metrics, streaks, and performance trends"
  guardrails:
    input:
      - type: pii-detector
        action: scrub
        fields: [name, email, phone, address]
    output:
      - type: toxicity-filter
        action: warn
        threshold: 0.7

---

Generated: agent.py

"""
fitness-tracker — LangGraph agent
Generated by agentspec generate --framework langgraph
"""
from __future__ import annotations

import os
from typing import Annotated, TypedDict
from urllib.parse import urlparse

from langchain_groq import ChatGroq
from langchain_core.messages import BaseMessage, HumanMessage, trim_messages
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

from guardrails import scrub_pii_input, check_content_safety_output
from agentspec import AgentSpecReporter


# ── Environment validation (always before any client init) ────────────────────

def validate_env() -> None:
    required = {"GROQ_API_KEY"}
    missing = [k for k in required if not os.environ.get(k)]
    if missing:
        raise EnvironmentError(
            f"Missing required environment variables: {', '.join(missing)}\n"
            "Set them in your .env file or deployment config."
        )


# ── Tools ─────────────────────────────────────────────────────────────────────

@tool
def plan_workout(
    goal: str,
    fitness_level: str,
    days_per_week: int,
) -> dict:
    """Generate a personalised workout plan based on user goals and fitness level."""
    # Implementation: call your workout planning service here
    return {
        "plan": f"Personalised {days_per_week}-day plan for {fitness_level} targeting {goal}",
        "exercises": [],
    }


@tool
def log_session(
    exercise: str,
    sets: int,
    reps: int,
    weight_kg: float,
) -> dict:
    """Record a completed workout session with exercises, sets, reps, and weight."""
    return {"logged": True, "exercise": exercise, "volume_kg": sets * reps * weight_kg}


@tool
def get_progress(
    user_id: str,
    metric: str = "strength",
) -> dict:
    """Retrieve user progress metrics, streaks, and performance trends."""
    return {"user_id": user_id, "metric": metric, "trend": "improving"}


TOOLS = [plan_workout, log_session, get_progress]


# ── Graph state ───────────────────────────────────────────────────────────────

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


# ── Graph nodes ───────────────────────────────────────────────────────────────

def call_model(state: AgentState, llm_with_tools: ChatGroq) -> AgentState:
    """Invoke the LLM with tool bindings."""
    messages = trim_messages(
        state["messages"],
        max_messages=20,
        max_tokens=4096,
        strategy="last",
        token_counter=llm_with_tools,
    )
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END


# ── Graph assembly ────────────────────────────────────────────────────────────

def build_graph(llm: ChatGroq) -> StateGraph:
    llm_with_tools = llm.bind_tools(TOOLS)
    from langgraph.prebuilt import ToolNode

    graph = StateGraph(AgentState)
    graph.add_node("agent", lambda s: call_model(s, llm_with_tools))
    graph.add_node("tools", ToolNode(TOOLS))
    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
    graph.add_edge("tools", "agent")
    return graph


# ── Guardrails ────────────────────────────────────────────────────────────────

def apply_input_guardrails(text: str) -> str:
    """Scrub PII from user input before sending to the model."""
    return scrub_pii_input(text, fields=["name", "email", "phone", "address"])


def apply_output_guardrails(text: str) -> str:
    """Warn on harmful or medical-advice content in model output."""
    return check_content_safety_output(text, categories=["harmful", "medical-advice"])


# ── AgentSpec SDK integration ─────────────────────────────────────────────────

def setup_reporter(manifest_path: str = "agent.yaml") -> AgentSpecReporter:
    """Wire up the AgentSpec SDK reporter for live health + push mode."""
    reporter = AgentSpecReporter.from_yaml(manifest_path)

    # Register tool handlers so the tool: checks pass in /agentspec/health
    reporter.register_tool("plan-workout", plan_workout)
    reporter.register_tool("log-session",  log_session)
    reporter.register_tool("get-progress", get_progress)

    return reporter


# ── Entry point ───────────────────────────────────────────────────────────────

def main() -> None:
    validate_env()

    llm = ChatGroq(
        model="llama-3.3-70b-versatile",
        api_key=os.environ["GROQ_API_KEY"],
        temperature=0,
    )
    graph = build_graph(llm).compile(checkpointer=MemorySaver())

    # Start AgentSpec SDK in push mode (reports health every 30s to the sidecar)
    reporter = setup_reporter()
    reporter.start_push_mode(interval_seconds=30)

    print("fitness-tracker agent ready. Type your message (Ctrl-C to exit).")
    config = {"configurable": {"thread_id": "demo-session"}}
    while True:
        user_input = input("You: ").strip()
        if not user_input:
            continue
        safe_input = apply_input_guardrails(user_input)
        result = graph.invoke({"messages": [HumanMessage(content=safe_input)]}, config)
        output = result["messages"][-1].content
        safe_output = apply_output_guardrails(output)
        print(f"Agent: {safe_output}")


if __name__ == "__main__":
    main()

---

Generated: guardrails.py

"""
Guardrail implementations for fitness-tracker.
Generated by agentspec generate --framework langgraph
"""
import re

_PII_PATTERNS = {
    "name":    re.compile(r"\b(?:my name is|i am|i'm)\s+\w+", re.IGNORECASE),
    "email":   re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+"),
    "phone":   re.compile(r"\b\d{3}[-.\s]?\d{3}[-.\s]?\d{4}\b"),
    "address": re.compile(r"\d+\s+\w+\s+(street|st|avenue|ave|road|rd|blvd)\b", re.IGNORECASE),
}

_HARMFUL_PATTERNS = [
    re.compile(r"\b(harm|hurt|injure|kill)\s+(yourself|myself)\b", re.IGNORECASE),
]
_MEDICAL_PATTERNS = [
    re.compile(r"\b(diagnos|prescri|medic[ai]|dosage|drug)\b", re.IGNORECASE),
]


def scrub_pii_input(text: str, fields: list[str]) -> str:
    for field in fields:
        if field in _PII_PATTERNS:
            text = _PII_PATTERNS[field].sub(f"[{field.upper()} REDACTED]", text)
    return text


def check_content_safety_output(text: str, categories: list[str]) -> str:
    if "harmful" in categories:
        for p in _HARMFUL_PATTERNS:
            if p.search(text):
                import warnings
                warnings.warn("Output flagged: harmful content detected", stacklevel=2)
    if "medical-advice" in categories:
        for p in _MEDICAL_PATTERNS:
            if p.search(text):
                import warnings
                warnings.warn("Output flagged: potential medical advice detected", stacklevel=2)
    return text

---

Generated: requirements.txt

langchain-groq>=0.2.0
langchain-core>=0.3.0
langgraph>=0.2.0
agentspec>=0.1.0
python-dotenv>=1.0.0

---

Generated: .env.example

# fitness-tracker — environment variables
# Copy to .env and fill in real values

GROQ_API_KEY=sk-...

---

How the SDK integration works

The generated code calls AgentSpecReporter.from_yaml("agent.yaml") and registers each tool handler. This enables:

Endpoint	Purpose
GET /agentspec/health	Live HealthReport — all 5 checks (env, model, 3 tools)
GET /health	Simple liveness check ({"status":"ok"})
GET /capabilities	Declared tools with live registration status

The sidecar detects /agentspec/health and switches to source=agent-sdk mode, giving you live-accurate compliance scoring instead of static analysis only.

Score breakdown (agent-sdk mode):

Check	Severity	Status	Impact
env:GROQ_API_KEY	error	pass	—
model:groq/llama-3.3	error	pass	—
tool:plan-workout	info	pass	—
tool:log-session	info	pass	—
tool:get-progress	info	pass	—
guardrails (audit)	—	pass	—

Score: 100 / Grade: A / Phase: Healthy