mirror/temporal-ai-agent
1
0
Fork 0
mirror of https://github.com/temporal-community/temporal-ai-agent.git synced 2026-03-16 22:48:09 +01:00
Code Issues Packages Projects Releases Wiki Activity

Enhance Dev Experience and Code Quality (#41)

Browse Source 
* Format codebase to satisfy linters

* fixing pylance and ruff-checked files

* contributing md, and type and formatting fixes

* setup file capitalization

* test fix
This commit is contained in:
Steve Androulakis
2025-06-01 08:54:59 -07:00
committed by GitHub
parent e35181b5ad
commit eb06cf5c8d
52 changed files with 1282 additions and 1105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tools/__init__.py
View File

@@ -1,29 +1,24 @@
from .search_fixtures import search_fixtures
from .search_flights import search_flights
from .search_trains import search_trains
from .search_trains import book_trains
from .create_invoice import create_invoice
from .find_events import find_events
from .list_agents import list_agents
from .change_goal import change_goal
from .transfer_control import transfer_control
from .hr.current_pto import current_pto
from .hr.book_pto import book_pto
from .hr.future_pto_calc import future_pto_calc
from .hr.checkpaybankstatus import checkpaybankstatus
from .create_invoice import create_invoice
from .ecommerce.get_order import get_order
from .ecommerce.list_orders import list_orders
from .ecommerce.track_package import track_package
from .fin.check_account_valid import check_account_valid
from .fin.get_account_balances import get_account_balance
from .fin.move_money import move_money
from .fin.submit_loan_application import submit_loan_application
from .ecommerce.get_order import get_order
from .ecommerce.track_package import track_package
from .ecommerce.list_orders import list_orders
from .find_events import find_events
from .give_hint import give_hint
from .guess_location import guess_location
from .hr.book_pto import book_pto
from .hr.checkpaybankstatus import checkpaybankstatus
from .hr.current_pto import current_pto
from .hr.future_pto_calc import future_pto_calc
from .list_agents import list_agents
from .search_fixtures import search_fixtures
from .search_flights import search_flights
from .search_trains import book_trains, search_trains
from .transfer_control import transfer_control
def get_handler(tool_name: str):

3
tools/change_goal.py
View File

@@ -1,9 +1,8 @@
def change_goal(args: dict) -> dict:
new_goal = args.get("goalID")
if new_goal is None:
new_goal = "goal_choose_agent_type"
return {
"new_goal": new_goal,
}
}

1
tools/create_invoice.py
View File

@@ -1,4 +1,5 @@
import os
import stripe
from dotenv import load_dotenv

14
tools/ecommerce/get_order.py
View File

@@ -1,16 +1,18 @@
from pathlib import Path
import json
from pathlib import Path
# this is made to demonstrate functionality but it could just as durably be an API call
# called as part of a temporal activity with automatic retries
def get_order(args: dict) -> dict:
order_id = args.get("order_id")
file_path = Path(__file__).resolve().parent.parent / "data" / "customer_order_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "customer_order_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
with open(file_path, "r") as file:
data = json.load(file)
order_list = data["orders"]
@@ -18,6 +20,6 @@ def get_order(args: dict) -> dict:
for order in order_list:
if order["id"] == order_id:
return order
return_msg = "Order " + order_id + " not found."
return {"error": return_msg}
return {"error": return_msg}

16
tools/ecommerce/list_orders.py
View File

@@ -1,17 +1,20 @@
from pathlib import Path
import json
from pathlib import Path
def sorting(e):
return e['order_date']
return e["order_date"]
def list_orders(args: dict) -> dict:
email_address = args.get("email_address")
file_path = Path(__file__).resolve().parent.parent / "data" / "customer_order_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "customer_order_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
with open(file_path, "r") as file:
data = json.load(file)
order_list = data["orders"]
@@ -24,7 +27,6 @@ def list_orders(args: dict) -> dict:
if len(rtn_order_list) > 0:
rtn_order_list.sort(key=sorting)
return {"orders": rtn_order_list}
else:
else:
return_msg = "No orders for customer " + email_address + " found."
return {"error": return_msg}

85
tools/ecommerce/track_package.py
View File

@@ -1,49 +1,59 @@
import http
import os
import json
import os
from pathlib import Path
#Send back dummy data in the correct format - to use the real API, 1) change this to be track_package_fake and 2) change the below track_package_real to be track_package
# Send back dummy data in the correct format - to use the real API, 1) change this to be track_package_fake and 2) change the below track_package_real to be track_package
def track_package(args: dict) -> dict:
tracking_id = args.get("tracking_id")
file_path = Path(__file__).resolve().parent.parent / "data" / "dummy_tracking_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "dummy_tracking_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
with open(file_path, "r") as file:
data = json.load(file)
package_list = data["packages"]
for package in package_list:
if package["TrackingNumber"] == tracking_id:
scheduled_delivery_date = package["ScheduledDeliveryDate"]
carrier = package["Carrier"]
status_summary = package["StatusSummary"]
tracking_details = package.get("TrackingDetails", [])
last_tracking_update = ""
if tracking_details and tracking_details is not None and tracking_details[0] is not None:
last_tracking_update = tracking_details[0]["EventDateTimeInDateTimeFormat"]
tracking_link = ""
if carrier == "USPS":
tracking_link = f"https://tools.usps.com/go/TrackConfirmAction?qtc_tLabels1={tracking_id}"
elif carrier == "UPS":
tracking_link = f"https://www.ups.com/track?track=yes&trackNums={tracking_id}"
scheduled_delivery_date = package["ScheduledDeliveryDate"]
carrier = package["Carrier"]
status_summary = package["StatusSummary"]
tracking_details = package.get("TrackingDetails", [])
last_tracking_update = ""
if (
tracking_details
and tracking_details is not None
and tracking_details[0] is not None
):
last_tracking_update = tracking_details[0][
"EventDateTimeInDateTimeFormat"
]
tracking_link = ""
if carrier == "USPS":
tracking_link = f"https://tools.usps.com/go/TrackConfirmAction?qtc_tLabels1={tracking_id}"
elif carrier == "UPS":
tracking_link = (
f"https://www.ups.com/track?track=yes&trackNums={tracking_id}"
)
return {
"scheduled_delivery_date": scheduled_delivery_date,
"carrier": carrier,
"status_summary": status_summary,
"tracking_link": tracking_link,
"last_tracking_update": last_tracking_update,
}
return {
"scheduled_delivery_date": scheduled_delivery_date,
"carrier": carrier,
"status_summary": status_summary,
"tracking_link": tracking_link,
"last_tracking_update": last_tracking_update
}
return_msg = "Package not found with tracking info " + tracking_id
return {"error": return_msg}
'''Format of response:
"""Format of response:
{
"TrackingNumber": "",
"Delivered": false,
@@ -94,9 +104,10 @@ def track_package(args: dict) -> dict:
}
]
}
'''
def track_package_real(args: dict) -> dict:
"""
def track_package_real(args: dict) -> dict:
tracking_id = args.get("tracking_id")
api_key = os.getenv("RAPIDAPI_KEY")
@@ -127,11 +138,17 @@ def track_package_real(args: dict) -> dict:
status_summary = json_data["StatusSummary"]
tracking_details = json_data.get("TrackingDetails", [])
last_tracking_update = ""
if tracking_details and tracking_details is not None and tracking_details[0] is not None:
if (
tracking_details
and tracking_details is not None
and tracking_details[0] is not None
):
last_tracking_update = tracking_details[0]["EventDateTimeInDateTimeFormat"]
tracking_link = ""
if carrier == "USPS":
tracking_link = f"https://tools.usps.com/go/TrackConfirmAction?qtc_tLabels1={tracking_id}"
tracking_link = (
f"https://tools.usps.com/go/TrackConfirmAction?qtc_tLabels1={tracking_id}"
)
elif carrier == "UPS":
tracking_link = f"https://www.ups.com/track?track=yes&trackNums={tracking_id}"
@@ -140,5 +157,5 @@ def track_package_real(args: dict) -> dict:
"carrier": carrier,
"status_summary": status_summary,
"tracking_link": tracking_link,
"last_tracking_update": last_tracking_update
}
"last_tracking_update": last_tracking_update,
}

23
tools/fin/check_account_valid.py
View File

@@ -1,24 +1,31 @@
from pathlib import Path
import json
from pathlib import Path
# this is made to demonstrate functionality but it could just as durably be an API call
# called as part of a temporal activity with automatic retries
def check_account_valid(args: dict) -> dict:
email = args.get("email")
account_id = args.get("account_id")
file_path = Path(__file__).resolve().parent.parent / "data" / "customer_account_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "customer_account_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
with open(file_path, "r") as file:
data = json.load(file)
account_list = data["accounts"]
for account in account_list:
if account["email"] == email or account["account_id"] == account_id:
return{"status": "account valid"}
return_msg = "Account not found with email address " + email + " or account ID: " + account_id
return {"error": return_msg}
return {"status": "account valid"}
return_msg = (
"Account not found with email address "
+ email
+ " or account ID: "
+ account_id
)
return {"error": return_msg}

24
tools/fin/get_account_balances.py
View File

@@ -1,23 +1,33 @@
from pathlib import Path
import json
from pathlib import Path
# this is made to demonstrate functionality but it could just as durably be an API call
# this assumes it's a valid account - use check_account_valid() to verify that first
def get_account_balance(args: dict) -> dict:
account_key = args.get("email_address_or_account_ID")
file_path = Path(__file__).resolve().parent.parent / "data" / "customer_account_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "customer_account_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
with open(file_path, "r") as file:
data = json.load(file)
account_list = data["accounts"]
for account in account_list:
if account["email"] == account_key or account["account_id"] == account_key:
return{ "name": account["name"], "email": account["email"], "account_id": account["account_id"], "checking_balance": account["checking_balance"], "savings_balance": account["savings_balance"], "bitcoin_balance": account["bitcoin_balance"], "account_creation_date": account["account_creation_date"] }
return {
"name": account["name"],
"email": account["email"],
"account_id": account["account_id"],
"checking_balance": account["checking_balance"],
"savings_balance": account["savings_balance"],
"bitcoin_balance": account["bitcoin_balance"],
"account_creation_date": account["account_creation_date"],
}
return_msg = "Account not found with for " + account_key
return {"error": return_msg}
return {"error": return_msg}

16
tools/fin/move_money.py
View File

@@ -1,16 +1,12 @@
import os
from pathlib import Path
import json
from temporalio.client import Client
import os
from dataclasses import dataclass
from typing import Optional
import asyncio
from pathlib import Path
from temporalio.exceptions import WorkflowAlreadyStartedError
from shared.config import get_temporal_client
from enum import Enum, auto
# enums for the java enum
# class ExecutionScenarios(Enum):
# HAPPY_PATH = 0
@@ -32,7 +28,6 @@ class MoneyMovementWorkflowParameterObj:
# this is made to demonstrate functionality but it could just as durably be an API call
# this assumes it's a valid account - use check_account_valid() to verify that first
async def move_money(args: dict) -> dict:
account_key = args.get("email_address_or_account_ID")
account_type: str = args.get("accounttype")
amount = args.get("amount")
@@ -101,7 +96,6 @@ async def move_money(args: dict) -> dict:
async def start_workflow(
amount_cents: int, from_account_name: str, to_account_name: str
) -> str:
start_real_workflow = os.getenv("FIN_START_REAL_WORKFLOW")
if start_real_workflow is not None and start_real_workflow.lower() == "false":
START_REAL_WORKFLOW = False
@@ -128,7 +122,7 @@ async def start_workflow(
task_queue="MoneyTransferJava", # Task queue name
)
return handle.id
except WorkflowAlreadyStartedError as e:
except WorkflowAlreadyStartedError:
existing_handle = client.get_workflow_handle(workflow_id=workflow_id)
return existing_handle.id
else:

76
tools/fin/submit_loan_application.py
View File

@@ -1,18 +1,10 @@
from datetime import date, timedelta
import os
from pathlib import Path
import json
from temporalio.client import (
Client,
WithStartWorkflowOperation,
WorkflowHandle,
WorkflowUpdateFailedError,
)
from temporalio import common
from dataclasses import dataclass
from typing import Optional
import asyncio
from temporalio.exceptions import WorkflowAlreadyStartedError
from datetime import date
from temporalio import common
from temporalio.client import WithStartWorkflowOperation, WorkflowUpdateFailedError
from shared.config import get_temporal_client
@@ -24,39 +16,55 @@ class TransactionRequest:
sourceAccount: str
targetAccount: str
@dataclass
class TxResult:
transactionId: str
status: str
#demonstrate starting a workflow and early return pattern while the workflow continues
# demonstrate starting a workflow and early return pattern while the workflow continues
async def submit_loan_application(args: dict) -> dict:
account_key = args.get("email_address_or_account_ID")
amount = args.get("amount")
loan_status: dict = await start_workflow(amount=amount,account_name=account_key)
loan_status: dict = await start_workflow(amount=amount, account_name=account_key)
if loan_status.get("error") is None:
return {'status': loan_status.get("loan_application_status"), 'detailed_status': loan_status.get("application_details"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("transaction_id")}
return {
"status": loan_status.get("loan_application_status"),
"detailed_status": loan_status.get("application_details"),
"next_step": loan_status.get("advisement"),
"confirmation_id": loan_status.get("transaction_id"),
}
else:
print(loan_status)
return loan_status
# Async function to start workflow
async def start_workflow(amount: str, account_name: str, )-> dict:
async def start_workflow(
amount: str,
account_name: str,
) -> dict:
start_real_workflow = os.getenv("FIN_START_REAL_WORKFLOW")
if start_real_workflow is not None and start_real_workflow.lower() == "false":
START_REAL_WORKFLOW = False
return {'loan_application_status': "applied", 'application_details': "loan application is submitted and initial validation is complete",'transaction_id': "APPLICATION"+account_name, 'advisement': "You'll receive a confirmation for final approval in three business days", }
# START_REAL_WORKFLOW = False
return {
"loan_application_status": "applied",
"application_details": "loan application is submitted and initial validation is complete",
"transaction_id": "APPLICATION" + account_name,
"advisement": "You'll receive a confirmation for final approval in three business days",
}
else:
START_REAL_WORKFLOW = True
# Connect to Temporal
# START_REAL_WORKFLOW = True
# Connect to Temporal
client = await get_temporal_client()
# Define the workflow ID and task queue
workflow_id = "LOAN_APPLICATION-"+account_name+"-"+date.today().strftime('%Y-%m-%d')
workflow_id = (
"LOAN_APPLICATION-" + account_name + "-" + date.today().strftime("%Y-%m-%d")
)
task_queue = "LatencyOptimizationTEST"
# Create a TransactionRequest (matching the Java workflow's expected input)
@@ -83,21 +91,27 @@ async def start_workflow(amount: str, account_name: str, )-> dict:
)
)
except WorkflowUpdateFailedError:
print("aww man got exception WorkflowUpdateFailedError" )
print("aww man got exception WorkflowUpdateFailedError")
tx_result = None
return_msg = "Loan could not be processed for " + account_name
return {"error": return_msg}
workflow_handle = await start_op.workflow_handle()
print(f"Workflow started with ID: {workflow_handle.id}")
print(tx_result)
print(f"Update result: Transaction ID = {tx_result.transactionId}, Message = {tx_result.status}")
print(
f"Update result: Transaction ID = {tx_result.transactionId}, Message = {tx_result.status}"
)
# Optionally, wait for the workflow to complete and get the final result
# final_result = await handle.result()
# print(f"Workflow completed with result: {final_result}")
# return {'status': loan_status.get("loan_status"), 'detailed_status': loan_status.get("results"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("workflowID")}
return {'loan_application_status': "applied", 'application_details': "loan application is submitted and initial validation is complete",'transaction_id': tx_result.transactionId, 'advisement': "You'll receive a confirmation for final approval in three business days", }
# return {'status': loan_status.get("loan_status"), 'detailed_status': loan_status.get("results"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("workflowID")}
return {
"loan_application_status": "applied",
"application_details": "loan application is submitted and initial validation is complete",
"transaction_id": tx_result.transactionId,
"advisement": "You'll receive a confirmation for final approval in three business days",
}

2
tools/find_events.py
View File

@@ -1,6 +1,6 @@
import json
from datetime import datetime
from pathlib import Path
import json
def find_events(args: dict) -> dict:

24
tools/give_hint.py
View File

@@ -1,10 +1,10 @@
TREASURE_LOCATION = {
"address": "300 Lenora",
"city": "Seattle",
"state_full": "Washington",
"state_abbrev": "WA",
"zip": "98121",
"country": "USA"
"address": "300 Lenora",
"city": "Seattle",
"state_full": "Washington",
"state_abbrev": "WA",
"zip": "98121",
"country": "USA",
}
HINTS = [
@@ -12,8 +12,8 @@ HINTS = [
"state of " + TREASURE_LOCATION["state_full"],
"city of " + TREASURE_LOCATION["city"],
"at a company HQ",
"The company's tech traces its roots to a project called Cadence", #thanks, Grok
"The company offers a tool that lets developers write code as if it's running forever, no matter what crashes", #thanks, Grok
"The company's tech traces its roots to a project called Cadence", # thanks, Grok
"The company offers a tool that lets developers write code as if it's running forever, no matter what crashes", # thanks, Grok
]
''' Additional Grok provided hints about Temporal:
"This company was founded by two engineers who previously worked on a system named after a South American river at Uber."
@@ -26,16 +26,14 @@ HINTS = [
"They’re backed by big venture capital names like Sequoia, betting on their vision for reliable software."
"The company’s name might remind you of a word for something fleeting, yet their tech is built to last."'''
def give_hint(args: dict) -> dict:
hint_total = args.get("hint_total")
if hint_total is None:
hint_total = 0
index = hint_total % len(HINTS)
hint_text = HINTS[index]
hint_total = hint_total + 1
return {
"hint_number": hint_total,
"hint": hint_text
}
return {"hint_number": hint_total, "hint": hint_text}

7
tools/goal_registry.py
View File

@@ -1,7 +1,8 @@
import os
from typing import List
from models.tool_definitions import AgentGoal
import tools.tool_registry as tool_registry
from models.tool_definitions import AgentGoal
# Turn on Silly Mode - this should be a description of the persona you'd like the bot to have and can be a single word or a phrase.
# Example if you want the bot to be a specific person, like Mario or Christopher Walken, or to describe a specific tone:
@@ -310,7 +311,7 @@ goal_fin_check_account_balances = AgentGoal(
)
# this tool checks account balances, and uses ./data/customer_account_data.json as dummy data
# it also uses a separate workflow/tool, see ./setup.md for details
# it also uses a separate workflow/tool, see ./SETUP.md for details
goal_fin_move_money = AgentGoal(
id="goal_fin_move_money",
category_tag="fin",
@@ -350,7 +351,7 @@ goal_fin_move_money = AgentGoal(
)
# this starts a loan approval process
# it also uses a separate workflow/tool, see ./setup.md for details
# it also uses a separate workflow/tool, see ./SETUP.md for details
goal_fin_loan_application = AgentGoal(
id="goal_fin_loan_application",
category_tag="fin",

12
tools/guess_location.py
View File

@@ -1,7 +1,7 @@
from .give_hint import TREASURE_LOCATION
def guess_location(args: dict) -> dict:
guess_address = args.get("address").lower()
guess_city = args.get("city").lower()
guess_state = args.get("state").lower()
@@ -11,8 +11,12 @@ def guess_location(args: dict) -> dict:
else:
compare_state = TREASURE_LOCATION.get("state_full").lower()
#Check for the street address to be included in the guess to account for "st" vs "street" or leaving Street off entirely
if TREASURE_LOCATION.get("address").lower() in guess_address and TREASURE_LOCATION.get("city").lower() == guess_city and compare_state == guess_state:
# Check for the street address to be included in the guess to account for "st" vs "street" or leaving Street off entirely
if (
TREASURE_LOCATION.get("address").lower() in guess_address
and TREASURE_LOCATION.get("city").lower() == guess_city
and compare_state == guess_state
):
return {"treasure_found": "True"}
else:
return {"treasure_found": "False"}
return {"treasure_found": "False"}

9
tools/hr/book_pto.py
View File

@@ -1,11 +1,10 @@
def book_pto(args: dict) -> dict:
email = args.get("email")
start_date = args.get("start_date")
end_date = args.get("end_date")
print(f"[BookPTO] Totally would send an email confirmation of PTO from {start_date} to {end_date} to {email} here!")
print(
f"[BookPTO] Totally would send an email confirmation of PTO from {start_date} to {end_date} to {email} here!"
)
return {
"status": "success"
}
return {"status": "success"}

7
tools/hr/checkpaybankstatus.py
View File

@@ -1,9 +1,4 @@
from pathlib import Path
import json
def checkpaybankstatus(args: dict) -> dict:
email = args.get("email")
if email == "grinch@grinch.com":
@@ -12,4 +7,4 @@ def checkpaybankstatus(args: dict) -> dict:
# could do logic here or look up data but for now everyone but the grinch is getting paid
return_msg = "connected"
return {"status": return_msg}
return {"status": return_msg}

15
tools/hr/current_pto.py
View File

@@ -1,26 +1,27 @@
from pathlib import Path
import json
from pathlib import Path
def current_pto(args: dict) -> dict:
email = args.get("email")
file_path = Path(__file__).resolve().parent.parent / "data" / "employee_pto_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "employee_pto_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
data = json.load(open(file_path))
employee_list = data["theCompany"]["employees"]
for employee in employee_list:
if employee["email"] == email:
num_hours = int(employee["currentPTOHrs"])
num_days = float(num_hours/8)
num_days = float(num_hours / 8)
return {
"num_hours": num_hours,
"num_days": num_days,
}
return_msg = "Employee not found with email address " + email
return {"error": return_msg}
return {"error": return_msg}

54
tools/hr/future_pto_calc.py
View File

@@ -1,43 +1,59 @@
import json
import pandas
from pathlib import Path
from datetime import date, datetime
from pathlib import Path
import pandas
from dateutil.relativedelta import relativedelta
def future_pto_calc(args: dict) -> dict:
file_path = Path(__file__).resolve().parent.parent / "data" / "employee_pto_data.json"
file_path = (
Path(__file__).resolve().parent.parent / "data" / "employee_pto_data.json"
)
if not file_path.exists():
return {"error": "Data file not found."}
start_date = datetime.strptime(args.get("start_date"), "%Y-%m-%d").date()
end_date = datetime.strptime(args.get("end_date"), "%Y-%m-%d").date()
email = args.get("email")
#Next, set up the ability to calculate how much PTO will be added to the user's total by the start of the PTO request
# Next, set up the ability to calculate how much PTO will be added to the user's total by the start of the PTO request
today = date.today()
if today > start_date:
return_msg = "PTO start date " + args.get("start_date") + "cannot be in the past"
return_msg = (
"PTO start date " + args.get("start_date") + "cannot be in the past"
)
return {"error": return_msg}
if end_date < start_date:
return_msg = "PTO end date " + args.get("end_date") + " must be after PTO start date " + args.get("start_date")
return_msg = (
"PTO end date "
+ args.get("end_date")
+ " must be after PTO start date "
+ args.get("start_date")
)
return {"error": return_msg}
#Get the number of business days, and then business hours (assume 8 hr biz day), included in the PTO request
biz_days_of_request = len(pandas.bdate_range(start=start_date, end=end_date, inclusive="both"))
# Get the number of business days, and then business hours (assume 8 hr biz day), included in the PTO request
biz_days_of_request = len(
pandas.bdate_range(start=start_date, end=end_date, inclusive="both")
)
if biz_days_of_request == 0:
return_msg = "There are no business days between " + args.get("start_date") + " and " + args.get("end_date")
return_msg = (
"There are no business days between "
+ args.get("start_date")
+ " and "
+ args.get("end_date")
)
return {"error": return_msg}
biz_hours_of_request = biz_days_of_request * 8
#Assume PTO is added on the first of every month - month math compares rolling dates, so compare the PTO request with the first day of the current month.
# Assume PTO is added on the first of every month - month math compares rolling dates, so compare the PTO request with the first day of the current month.
today_first_of_month = date(today.year, today.month, 1)
time_difference = relativedelta(start_date, today_first_of_month)
months_to_accrue = time_difference.years * 12 + time_difference.months
data = json.load(open(file_path))
employee_list = data["theCompany"]["employees"]
@@ -47,12 +63,14 @@ def future_pto_calc(args: dict) -> dict:
if employee["email"] == email:
current_pto_hours = int(employee["currentPTOHrs"])
hrs_added_per_month = int(employee["hrsAddedPerMonth"])
pto_available_at_start = current_pto_hours + (months_to_accrue * hrs_added_per_month)
pto_available_at_start = current_pto_hours + (
months_to_accrue * hrs_added_per_month
)
pto_hrs_remaining_after = pto_available_at_start - biz_hours_of_request
if pto_hrs_remaining_after >= 0:
enough_pto = True
return {
"enough_pto": enough_pto,
"enough_pto": enough_pto,
"pto_hrs_remaining_after": str(pto_hrs_remaining_after),
}

18
tools/list_agents.py
View File

@@ -1,19 +1,23 @@
import os
import tools.goal_registry as goals
def list_agents(args: dict) -> dict:
def list_agents(args: dict) -> dict:
goal_categories_start = os.getenv("GOAL_CATEGORIES")
if goal_categories_start is None:
goal_categories = ["all"] # default to 'all' categories
goal_categories = ["all"] # default to 'all' categories
else:
goal_categories_start.strip().lower() # handle extra spaces or non-lowercase
goal_categories_start.strip().lower() # handle extra spaces or non-lowercase
goal_categories = goal_categories_start.split(",")
# if multi-goal-mode, add agent_selection as a goal (defaults to True)
if "agent_selection" not in goal_categories :
first_goal_value = os.getenv("AGENT_GOAL")
if first_goal_value is None or first_goal_value.lower() == "goal_choose_agent_type":
if "agent_selection" not in goal_categories:
first_goal_value = os.getenv("AGENT_GOAL")
if (
first_goal_value is None
or first_goal_value.lower() == "goal_choose_agent_type"
):
goal_categories.append("agent_selection")
# always show goals labeled as "system," like the goal chooser
@@ -33,7 +37,7 @@ def list_agents(args: dict) -> dict:
"goal_id": goal.id,
"agent_description": goal.agent_friendly_description,
}
)
)
return {
"agents": agents,
}

5
tools/search_fixtures.py
View File

@@ -1,7 +1,8 @@
import os
import requests
import random
from datetime import datetime, timedelta, date
from datetime import date, datetime, timedelta
import requests
from dotenv import load_dotenv
PREMIER_LEAGUE_CLUBS_DATA = [

7
tools/search_flights.py
View File

@@ -1,9 +1,10 @@
import os
import json
import http.client
from dotenv import load_dotenv
import json
import os
import urllib.parse
from dotenv import load_dotenv
def search_airport(query: str) -> list:
"""

2
tools/tool_registry.py
View File

@@ -1,4 +1,4 @@
from models.tool_definitions import ToolDefinition, ToolArgument
from models.tool_definitions import ToolArgument, ToolDefinition
# ----- System tools -----
list_agents_tool = ToolDefinition(

4
tools/transfer_control.py
View File

@@ -1,7 +1,7 @@
import shared.config
def transfer_control(args: dict) -> dict:
def transfer_control(args: dict) -> dict:
return {
"new_goal": shared.config.AGENT_GOAL,
}
}