FCCS MCP Agentic Server

Instructions

Implementation Reference

• fccs_agent/tools/jobs.py:44-58 (handler)

  The main asynchronous handler function that executes the run_business_rule tool logic by calling the underlying FCCS client.

  async def run_business_rule(
      rule_name: str,
      parameters: Optional[dict[str, Any]] = None
  ) -> dict[str, Any]:
      """Run a business rule (e.g., Consolidation) / Executar regra de negocio.
  
      Args:
          rule_name: The name of the business rule to run.
          parameters: Optional parameters for the rule.
  
      Returns:
          dict: Job submission result.
      """
      result = await _client.run_business_rule(_app_name, rule_name, parameters)
      return {"status": "success", "data": result}

• fccs_agent/tools/jobs.py:109-126 (schema)

  The input schema definition for the run_business_rule tool, defining parameters rule_name (required) and optional parameters.

  {
      "name": "run_business_rule",
      "description": "Run a business rule (e.g., Consolidation) / Executar regra de negocio",
      "inputSchema": {
          "type": "object",
          "properties": {
              "rule_name": {
                  "type": "string",
                  "description": "The name of the business rule to run",
              },
              "parameters": {
                  "type": "object",
                  "description": "Optional parameters for the rule",
              },
          },
          "required": ["rule_name"],
      },
  },

• fccs_agent/agent.py:138-185 (registration)

  Registration of tool handlers including run_business_rule in the agent's TOOL_HANDLERS dictionary, used by execute_tool to dispatch tool calls.

  TOOL_HANDLERS = {
      # Application
      "get_application_info": application.get_application_info,
      "get_rest_api_version": application.get_rest_api_version,
      # Jobs
      "list_jobs": jobs.list_jobs,
      "get_job_status": jobs.get_job_status,
      "run_business_rule": jobs.run_business_rule,
      "run_data_rule": jobs.run_data_rule,
      # Dimensions
      "get_dimensions": dimensions.get_dimensions,
      "get_members": dimensions.get_members,
      "get_dimension_hierarchy": dimensions.get_dimension_hierarchy,
      # Journals
      "get_journals": journals.get_journals,
      "get_journal_details": journals.get_journal_details,
      "perform_journal_action": journals.perform_journal_action,
      "update_journal_period": journals.update_journal_period,
      "export_journals": journals.export_journals,
      "import_journals": journals.import_journals,
      # Data
      "export_data_slice": data.export_data_slice,
      "smart_retrieve": data.smart_retrieve,
      "smart_retrieve_consolidation_breakdown": data.smart_retrieve_consolidation_breakdown,
      "smart_retrieve_with_movement": data.smart_retrieve_with_movement,
      "copy_data": data.copy_data,
      "clear_data": data.clear_data,
      # Reports
      "generate_report": reports.generate_report,
      "get_report_job_status": reports.get_report_job_status,
      "generate_report_script": reports.generate_report_script,
      # Consolidation
      "export_consolidation_rulesets": consolidation.export_consolidation_rulesets,
      "import_consolidation_rulesets": consolidation.import_consolidation_rulesets,
      "validate_metadata": consolidation.validate_metadata,
      "generate_intercompany_matching_report": consolidation.generate_intercompany_matching_report,
      "import_supplementation_data": consolidation.import_supplementation_data,
      "deploy_form_template": consolidation.deploy_form_template,
      "generate_consolidation_process_report": consolidation.generate_consolidation_process_report,
      # Memo
      "generate_system_pitch": memo.generate_system_pitch,
      "generate_investment_memo": memo.generate_investment_memo,
      # Feedback
      "submit_feedback": feedback.submit_feedback,
      "get_recent_executions": feedback.get_recent_executions,
      # Local Data
      "query_local_metadata": local_data.query_local_metadata,
  }

• fccs_agent/agent.py:188-199 (registration)

  Collection of all tool definitions including the run_business_rule schema from jobs.TOOL_DEFINITIONS for MCP tool discovery.

  ALL_TOOL_DEFINITIONS = (
      application.TOOL_DEFINITIONS +
      jobs.TOOL_DEFINITIONS +
      dimensions.TOOL_DEFINITIONS +
      journals.TOOL_DEFINITIONS +
      data.TOOL_DEFINITIONS +
      reports.TOOL_DEFINITIONS +
      consolidation.TOOL_DEFINITIONS +
      memo.TOOL_DEFINITIONS +
      feedback.TOOL_DEFINITIONS +
      local_data.TOOL_DEFINITIONS
  )

• fccs_agent/client/fccs_client.py:58-146 (helper)

  Underlying client method called by the tool handler to interact with FCCS REST API for running business rules.

          self._fcm_client = httpx.AsyncClient(
              base_url=fcm_base_url,
              headers=headers,
              timeout=60.0,
          )
  
  async def close(self):
      """Close HTTP clients."""
      if self._client:
          await self._client.aclose()
      if self._fcm_client:
          await self._fcm_client.aclose()
  
  def _get_query_params(self, has_existing_query: bool = False) -> str:
      """Get admin mode query parameter if needed."""
      if not self.admin_mode:
          return ""
      return "&adminMode=true" if has_existing_query else "?adminMode=true"
  
  # ========== Application Methods ==========
  
  async def get_applications(self) -> dict[str, Any]:
      """Get FCCS applications / Obter aplicacoes FCCS."""
      if self.config.fccs_mock_mode:
          return MOCK_APPLICATIONS
  
      response = await self._client.get("/")
      response.raise_for_status()
      data = response.json()
  
      # Check if application is in admin mode
      if data.get("items") and len(data["items"]) > 0:
          if data["items"][0].get("adminMode"):
              self.admin_mode = True
  
      return data
  
  async def get_rest_api_version(self) -> dict[str, Any]:
      """Get REST API version / Obter versao da API REST."""
      if self.config.fccs_mock_mode:
          return {"version": self.config.fccs_api_version, "apiVersion": "3.0"}
  
      # Try version endpoints
      for endpoint in ["/rest/version", "/version", "/api/version"]:
          try:
              response = await self._client.get(endpoint)
              if response.status_code == 200:
                  return response.json()
          except Exception:
              continue
  
      return {
          "version": self.config.fccs_api_version,
          "note": "Version endpoint not available, using configured version"
      }
  
  # ========== Job Methods ==========
  
  async def list_jobs(self, app_name: str) -> dict[str, Any]:
      """List jobs / Listar trabalhos."""
      if self.config.fccs_mock_mode:
          return MOCK_JOBS
  
      try:
          response = await self._client.get(
              f"/{app_name}/jobs{self._get_query_params()}"
          )
          if response.status_code == 200:
              return response.json()
          return {"items": []}
      except Exception as e:
          return {"items": [], "error": str(e)}
  
  async def get_job_status(self, app_name: str, job_id: str) -> dict[str, Any]:
      """Get job status / Obter status do trabalho."""
      if self.config.fccs_mock_mode:
          return MOCK_JOB_STATUS.get(
              job_id,
              {"jobId": job_id, "status": "Unknown", "details": "Mock job not found"}
          )
  
      response = await self._client.get(
          f"/{app_name}/jobs/{job_id}{self._get_query_params()}"
      )
      response.raise_for_status()
      return response.json()
  
  async def run_business_rule(