harvesthealth's picture
Upload folder using huggingface_hub
c2ebec8 verified
import requests
import os
import json
JULES_API_ENDPOINT = "https://jules.googleapis.com/v1alpha"
def list_sources(api_key):
"""
Lists all available sources (e.g., GitHub repos) connected to Jules.
Args:
api_key (str): The Jules API key for authentication.
Returns:
dict: The JSON response from the API, or an error message.
"""
url = f'{JULES_API_ENDPOINT}/sources'
headers = {'X-Goog-Api-Key': api_key}
try:
response = requests.get(url, headers=headers, timeout=15)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {'error': f'Exception during request: {str(e)}'}
def create_session(api_key, source_name, prompt, starting_branch="main", title="Jules API Session"):
"""
Creates a new session in Jules.
Args:
api_key (str): The Jules API key for authentication.
source_name (str): The name of the source to work with (e.g., 'sources/github/owner/repo').
prompt (str): The initial prompt for the agent.
starting_branch (str, optional): The branch to start from. Defaults to "main".
title (str, optional): The title for the session. Defaults to "Jules API Session".
Returns:
dict: The JSON response from the API, or an error message.
"""
url = f'{JULES_API_ENDPOINT}/sessions'
headers = {
'X-Goog-Api-Key': api_key,
'Content-Type': 'application/json'
}
data = {
"prompt": prompt,
"sourceContext": {
"source": source_name,
"githubRepoContext": {
"startingBranch": starting_branch
}
},
"automationMode": "AUTO_CREATE_PR",
"title": title
}
try:
response = requests.post(url, headers=headers, json=data, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {'error': f'Exception during request: {str(e)}'}
def list_session_activities(api_key, session_id):
"""
Lists all activities for a given Jules session.
Args:
api_key (str): The Jules API key for authentication.
session_id (str): The ID of the session to query.
Returns:
dict: The JSON response from the API, or an error message.
"""
url = f'{JULES_API_ENDPOINT}/sessions/{session_id}/activities?pageSize=30'
headers = {
'X-Goog-Api-Key': api_key,
'Content-Type': 'application/json'
}
try:
response = requests.get(url, headers=headers, timeout=15)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {'error': f'Exception during request: {str(e)}'}
def send_message_to_session(api_key, session_id, message):
"""
Sends a message to a Jules session.
Args:
api_key (str): The Jules API key for authentication.
session_id (str): The ID of the session to send the message to.
message (str): The message content.
Returns:
dict: The JSON response from the API, or an error message.
"""
url = f'{JULES_API_ENDPOINT}/sessions/{session_id}:sendMessage'
headers = {
'X-Goog-Api-Key': api_key,
'Content-Type': 'application/json'
}
data = {'prompt': message}
try:
response = requests.post(url, headers=headers, json=data, timeout=15)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {'error': f'Exception during request: {str(e)}'}
if __name__ == '__main__':
# This is an example of how to use the functions.
# You must set the JULES_API_KEY environment variable.
jules_key = os.environ.get("JULES_API_KEY")
if not jules_key:
print("Error: JULES_API_KEY environment variable not set.")
else:
print("--- Listing Sources ---")
sources = list_sources(jules_key)
print(json.dumps(sources, indent=2))
if sources and 'sources' in sources and len(sources['sources']) > 0:
# Use the first available source for the example
first_source = sources['sources'][0]['name']
print(f"\n--- Creating Session on source: {first_source} ---")
example_prompt = "Please add a new function to the main library file that calculates the factorial of a number."
session_info = create_session(jules_key, first_source, example_prompt, title="Factorial Function Task")
print(json.dumps(session_info, indent=2))
else:
print("\nNo sources found. Cannot create a session.")