get_journals

Instructions

Implementation Reference

• fccs_agent/tools/journals.py:21-60 (handler)

  The primary handler function for the 'get_journals' tool. It accepts input parameters, constructs filters, invokes the FCCS client method, and formats the response.

  async def get_journals(
      scenario: str,
      year: str,
      period: str,
      consolidation: str = "FCCS_Entity Input",
      view: str = "FCCS_Periodic",
      status: str = "Posted",
      offset: int = 0,
      limit: int = 100
  ) -> dict[str, Any]:
      """Retrieve consolidation journals with filters / Obter diarios de consolidacao.
  
      Note: FCCS REST API requires scenario, year, period, consolidation, view, and status.
  
      Args:
          scenario: Filter by scenario (required, e.g., 'Actual').
          year: Filter by year (required, e.g., 'FY25').
          period: Filter by period (required, e.g., 'Jan', 'Feb', 'Mar').
          consolidation: Filter by consolidation member (default: 'FCCS_Entity Input').
          view: Filter by view (default: 'FCCS_Periodic').
          status: Filter by status (default: 'Posted'). Valid: 'Working', 'Submitted', 'Approved', 'Rejected', 'Posted'.
          offset: Offset for pagination (default: 0).
          limit: Limit results (default: 100).
  
      Returns:
          dict: List of journals.
      """
      filters = {
          "scenario": scenario,
          "year": year,
          "period": period,
          "consolidation": consolidation,
          "view": view,
          "status": status,
      }
  
      journals = await _client.get_journals(
          _app_name, filters, offset, limit
      )
      return {"status": "success", "data": journals}

• fccs_agent/tools/journals.py:170-186 (schema)

  The input schema and metadata definition for the 'get_journals' tool, including properties, descriptions, defaults, and required fields.

      "name": "get_journals",
      "description": "Retrieve consolidation journals with optional filters / Obter diarios de consolidacao",
      "inputSchema": {
          "type": "object",
          "properties": {
              "scenario": {"type": "string", "description": "Filter by scenario (required, e.g., 'Actual')"},
              "year": {"type": "string", "description": "Filter by year (required, e.g., 'FY25')"},
              "period": {"type": "string", "description": "Filter by period (required, e.g., 'Jan', 'Feb', 'Mar')"},
              "consolidation": {"type": "string", "description": "Filter by consolidation member (default: 'FCCS_Entity Input')", "default": "FCCS_Entity Input"},
              "view": {"type": "string", "description": "Filter by view (default: 'FCCS_Periodic')", "default": "FCCS_Periodic"},
              "status": {"type": "string", "description": "Filter by status (default: 'Posted'). Valid: 'Working', 'Submitted', 'Approved', 'Rejected', 'Posted'", "default": "Posted"},
              "offset": {"type": "number", "description": "Offset for pagination (default: 0)"},
              "limit": {"type": "number", "description": "Limit results (default: 100)"},
          },
          "required": ["scenario", "year", "period"],
      },
  },

• fccs_agent/agent.py:152-152 (registration)

  Maps the tool name 'get_journals' to its handler function in the central TOOL_HANDLERS registry used by the agent.

  "get_journals": journals.get_journals,

• fccs_agent/client/fccs_client.py:381-426 (helper)

  The low-level FCCSClient method that performs the actual API call to retrieve journals, used by the tool handler.

  async def get_journals(
      self,
      app_name: str,
      filters: Optional[dict[str, str]] = None,
      offset: int = 0,
      limit: int = 100
  ) -> dict[str, Any]:
      """Get journals / Obter lancamentos."""
      if self.config.fccs_mock_mode:
          return {"items": [], "offset": offset, "limit": limit, "count": 0}
  
      # Build query parameters
      params = {"offset": offset, "limit": limit}
      
      if filters:
          # Build the q parameter as a JSON-like string
          filter_parts = []
          for key in ["scenario", "year", "period", "status", "consolidation", "view", "entity", "group", "label", "description"]:
              if key in filters:
                  filter_parts.append(f'"{key}":"{filters[key]}"')
          if filter_parts:
              # Use proper JSON format for the q parameter
              params["q"] = "{" + ",".join(filter_parts) + "}"
  
      # Use httpx params to handle URL encoding automatically
      response = await self._client.get(
          f"/{app_name}/journals",
          params=params
      )
  
      if response.status_code == 200:
          data = response.json()
          if not data.get("items"):
              data["message"] = "No journals currently exist in the system."
          return data
  
      # Return error details if not 200
      return {
          "items": [],
          "offset": offset,
          "limit": limit,
          "count": 0,
          "error": f"HTTP {response.status_code}",
          "detail": response.text[:500] if response.text else "No response body"
      }