Latest Goose API

main.py

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import subprocess
import re
import os


# Clean the output of the goose command
def clean_goose_output(output: str) -> str:
    # Remove ANSI color codes
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    clean_output = ansi_escape.sub('', output)

    # Remove lines about logging and directories
    filtered_lines = []
    for line in clean_output.splitlines():
        if not any(kw in line.lower() for kw in ["logging to", "working directory", "starting session"]):
            filtered_lines.append(line.strip())

    return "\n".join(filtered_lines).strip()


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class GooseInput(BaseModel):
    instructions: str
    session_name: str = "api-session"

@app.post("/run-goose/")
def run_goose(input: GooseInput):
    try:
        instruction_file = "/tmp/goose_input.txt"
        with open(instruction_file, "w") as f:
            f.write(input.instructions)

        env = os.environ.copy()
        env["GOOSE_NO_KEYRING"] = "1"
        env["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=/run/user/0/bus"
        env["RUST_BACKTRACE"] = "1"
        
        try:
            result = subprocess.run(
                ["goose", "run", "-i", instruction_file],
                capture_output=True,
                text=True,
                env=env
            )

            if result.returncode != 0:
                error_msg = result.stderr if result.stderr else result.stdout
                raise HTTPException(status_code=500, detail=error_msg)

            cleaned_stdout = clean_goose_output(result.stdout)
            return {"response": cleaned_stdout}

        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))