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(