from typing import List, Union, Dict from pydantic import BaseModel, Field import requests class Pipe: class Valves(BaseModel): API_KEY: str = Field(default="your_letta_api_key") AGENT_ID: str = Field(default="your_agent_id") BASE_URL: str = Field(default="https://api.letta.com") def __init__(self): self.type = "manifold" self.id = "letta-agents" self.name = "letta-agents/" self.valves = self.Valves() self.conversation_id = None def get_model_ids(self) -> List[str]: return ["letta-agent"] def pipes(self) -> List[Dict[str, str]]: return [{"id": m, "name": f"{self.name}{m}"} for m in self.get_model_ids()] def pipe(self, body: dict) -> Union[str, dict]: msgs = body.get("messages", []) if not msgs: return "Error: no messages passed" user_msg = next( (m["content"] for m in reversed(msgs) if m.get("role") == "user"), "" ) if not user_msg: return "Error: no valid user message found" try: url = f"{self.valves.BASE_URL.rstrip('/')}/v1/agents/{self.valves.AGENT_ID}/messages" print("DEBUG URL:", url) headers = { "Authorization": f"Bearer {self.valves.API_KEY}", "Content-Type": "application/json", } payload = {"messages": [{"role": "user", "content": user_msg}]} res = requests.post(url, headers=headers, json=payload) res.raise_for_status() data = res.json() # response "messages" is a list content = data.get("messages", [{}])[-1].get("content", "(no response)") return {"choices": [{"message": {"role": "assistant", "content": content}}]} except Exception as e: print("ERROR", e) return f"Error: {e}"