Repository navigation
Expand file tree
/
Copy pathazure_python_code_interpreter.py
More file actions
70 lines (51 loc) · 2.45 KB
/
Copy pathazure_python_code_interpreter.py
File metadata and controls
70 lines (51 loc) · 2.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import datetime
from azure.core.credentials import AccessToken
from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential
from semantic_kernel.connectors.ai.open_ai.services.azure_chat_completion import AzureChatCompletion
from semantic_kernel.core_plugins.sessions_python_tool.sessions_python_plugin import (
SessionsPythonTool,
)
from semantic_kernel.exceptions.function_exceptions import FunctionExecutionException
from semantic_kernel.kernel import Kernel
from semantic_kernel.utils.settings import (
azure_container_apps_settings_from_dot_env_as_dict,
azure_openai_settings_from_dot_env_as_dict,
)
auth_token: AccessToken | None = None
ACA_TOKEN_ENDPOINT = "https://acasessions.io/.default"
async def auth_callback() -> str:
"""Auth callback for the SessionsPythonTool.
This is a sample auth callback that shows how to use Azure's DefaultAzureCredential
to get an access token.
"""
global auth_token
current_utc_timestamp = int(datetime.datetime.now(datetime.timezone.utc).timestamp())
if not auth_token or auth_token.expires_on < current_utc_timestamp:
credential = DefaultAzureCredential()
try:
auth_token = credential.get_token(ACA_TOKEN_ENDPOINT)
except ClientAuthenticationError as cae:
err_messages = getattr(cae, "messages", [])
raise FunctionExecutionException(
f"Failed to retrieve the client auth token with messages: {' '.join(err_messages)}"
) from cae
return auth_token.token
async def main():
kernel = Kernel()
service_id = "python-code-interpreter"
chat_service = AzureChatCompletion(
service_id=service_id, **azure_openai_settings_from_dot_env_as_dict(include_api_version=True)
)
kernel.add_service(chat_service)
python_code_interpreter = SessionsPythonTool(
**azure_container_apps_settings_from_dot_env_as_dict(), auth_callback=auth_callback
)
sessions_tool = kernel.add_plugin(python_code_interpreter, "PythonCodeInterpreter")
code = "import json\n\ndef add_numbers(a, b):\n return a + b\n\nargs = '{\"a\": 1, \"b\": 1}'\nargs_dict = json.loads(args)\nprint(add_numbers(args_dict['a'], args_dict['b']))" # noqa: E501
result = await kernel.invoke(sessions_tool["execute_code"], code=code)
print(result)
if __name__ == "__main__":
asyncio.run(main())