File size: 1,371 Bytes
079ae3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | # envs/finqa_env/server/app.py
"""
FastAPI server for the FinQA environment.
Environment Variables:
FINQA_DATA_PATH: Path to data directory (default: /app/env/data)
FINQA_MAX_STEPS: Maximum tool calls per episode (default: 50)
FINQA_TASK: Task name (default: finqa)
"""
import json
import os
from typing import Any, Dict
from openenv.core.env_server.http_server import create_app
from openenv.core.env_server.mcp_types import CallToolAction, CallToolObservation
from pydantic import field_validator
from .finqa_environment import FinQAEnvironment
DATA_PATH = os.environ.get("FINQA_DATA_PATH", "/app/env/data")
MAX_STEPS = int(os.environ.get("FINQA_MAX_STEPS", "50"))
TASK = os.environ.get("FINQA_TASK", "finqa")
def _env_factory():
"""Create a new FinQAEnvironment instance for each session."""
return FinQAEnvironment(
data_path=DATA_PATH,
max_steps=MAX_STEPS,
task=TASK,
)
class FinQACallToolAction(CallToolAction):
"""CallToolAction that accepts JSON strings for arguments (web UI sends strings)."""
@field_validator("arguments", mode="before")
@classmethod
def parse_arguments(cls, v: Any) -> Dict[str, Any]:
if isinstance(v, str):
return json.loads(v)
return v
app = create_app(
_env_factory, FinQACallToolAction, CallToolObservation, env_name="finqa_env"
)
|