fetch_employees
Retrieve employee data from Paylocity for a specified company. Use this tool to access employee records, manage workforce information, and integrate with HR systems.
Instructions
Fetch all employees for a company.
Args: company_id: Optional company ID (string or integer). If not provided, uses the first company ID from configuration.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| company_id | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mcppaylocity/__init__.py:234-243 (handler)The primary handler for the 'fetch_employees' tool. Registered with @mcp.tool() decorator. Handles optional company_id parameter, selects default from config if omitted, applies retry via with_retry helper, and calls PaylocityClient.get_all_employees for the actual data fetch.
def fetch_employees(company_id: Optional[Union[str, int]] = None) -> Dict[str, Any]: """ Fetch all employees for a company. Args: company_id: Optional company ID (string or integer). If not provided, uses the first company ID from configuration. """ company_id_str = str(company_id) if company_id is not None else company_ids[0] return with_retry(client.get_all_employees, company_id_str) - src/mcppaylocity/__init__.py:217-232 (helper)Helper function implementing retry logic with exponential backoff, used by the fetch_employees tool (and others) to handle transient API failures.
def with_retry(func, *args, **kwargs): max_retries = 3 retry_delay = 1 # seconds for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt < max_retries - 1: logger.warning("Attempt %d failed: %s. Retrying in %d seconds...", attempt+1, str(e), retry_delay) time.sleep(retry_delay) retry_delay *= 2 # Exponential backoff else: logger.error("Failed after %d attempts: %s", max_retries, str(e)) raise - Supporting method in PaylocityClient class that performs the actual HTTP API call to Paylocity's /employees endpoint, with pagination parameters and automatic token handling via _make_request.
def get_all_employees(self, company_id) -> Dict[str, Any]: """Get all employees with automatic token management""" endpoint = "/api/v2/companies/{}/employees".format(company_id) params = { "pagesize": 100, "pagenumber": 0, "includetotalcount": True } try: return self._make_request("GET", endpoint, params=params).json() except Exception as e: logger.error("Failed to get employees for company %s: %s", company_id, str(e)) raise