Microsoft Foundry connectors

1
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
2
AZURE_AI_API_KEY=your_azure_openai_api_key
3
AZURE_AI_DEPLOYMENT_NAME=gpt-4o

1
# Fix SSL certificate verification (must be before other imports)
2
import truststore
3
truststore.inject_into_ssl()
4

5
import os
6

7
from agent_framework.openai import OpenAIChatClient
8
from dotenv import load_dotenv
9
from pathlib import Path
10

11
from tac_microsoft import (
12
    TAC,
13
    TACConfig,
14
    SMSChannelConfig,
15
    TACFastAPIServer,
16
    AgentFrameworkConnector,
17
    ConversationSession,
18
)
19

20
load_dotenv(Path(__file__).resolve().parent.parent / ".env")
21

22
# Azure AI client
23
client = OpenAIChatClient(
24
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
25
    api_key=os.environ["AZURE_AI_API_KEY"],
26
    model=os.environ.get("AZURE_AI_DEPLOYMENT_NAME"),
27
)
28

29
# TAC setup
30
tac = TAC(config=TACConfig.from_env())
31

32
SYSTEM_PROMPT = "You are a helpful customer service agent. Be concise and friendly."
33

34

35
def create_agent(session: ConversationSession):
36
    """Return an Agent Framework agent for this conversation."""
37
    return client.as_agent(
38
        name="MyAgent",
39
        instructions=SYSTEM_PROMPT,
40
    )
41

42

43
connector = AgentFrameworkConnector(
44
    tac=tac,
45
    create_agent=create_agent,
46
    sms_config=SMSChannelConfig(memory_mode="always"),
47
)
48

49
server = TACFastAPIServer(
50
    tac=tac,
51
    voice_channel=connector.voice_channel,
52
    messaging_channels=[connector.sms_channel],
53
)
54

55
if __name__ == "__main__":
56
    server.start()

1
# Fix SSL certificate verification (must be before other imports)
2
import truststore
3
truststore.inject_into_ssl()
4

5
import asyncio
6
import logging
7
import os
8
from pathlib import Path
9

10
from agent_framework.openai import OpenAIChatClient
11
from azure.identity.aio import DefaultAzureCredential
12
from dotenv import load_dotenv
13

14
from tac_microsoft import (
15
    TAC,
16
    TACConfig,
17
    TACFastAPIServer,
18
    AgentFrameworkConnector,
19
    FileAgentSessionStore,
20
    ConversationSession,
21
    VoiceChannelConfig,
22
    SMSChannelConfig,
23
    format_memory_context,
24
)
25
from tac_microsoft.agent_framework_tools import create_knowledge_tool, create_memory_tool
26

27
load_dotenv(Path(__file__).resolve().parent.parent / ".env")
28

29
logger = logging.getLogger(__name__)
30

31
credential = DefaultAzureCredential()
32
client = OpenAIChatClient(
33
    credential=credential,
34
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
35
    model=os.environ.get("AZURE_AI_DEPLOYMENT_NAME", "gpt-4o"),
36
)
37

38
tac = TAC(config=TACConfig.from_env())
39
knowledge_base_id = os.environ.get("TWILIO_KNOWLEDGE_BASE_ID")
40

41
# Build the knowledge tool once at startup — it doesn't depend on session state.
42
knowledge_tool = (
43
    asyncio.run(create_knowledge_tool(
44
        tac,
45
        knowledge_base_id,
46
        description="Search the knowledge base for relevant information.",
47
    ))
48
    if knowledge_base_id
49
    else None
50
)
51

52
# Channel-aware system prompts
53
VOICE_SYSTEM_PROMPT = """You are a customer service assistant on a phone call.
54
Keep responses clear, concise, and conversational.
55
Use plain text only — no markdown, no lists, no special formatting."""
56

57
SMS_SYSTEM_PROMPT = """You are a customer service assistant over SMS.
58
Keep responses concise and formatted for text messaging.
59
Use short paragraphs. Bullet points are OK."""
60

61

62
# Custom tool
63
def look_up_outage_tool(zip_code: str) -> str:
64
    """Check if there is a recent internet outage in a specific zip code."""
65
    return f"No reported outages in {zip_code}. Service is operating normally."
66

67

68
# Agent factory — called once per voice call, once per SMS message
69
def create_agent(session: ConversationSession):
70
    prompt = VOICE_SYSTEM_PROMPT if session.channel == "voice" else SMS_SYSTEM_PROMPT
71

72
    tools = [create_memory_tool(tac, session), look_up_outage_tool, knowledge_tool]
73

74
    return client.as_agent(
75
        name="OwlAgent",
76
        instructions=prompt,
77
        tools=[t for t in tools if t is not None],
78
    )
79

80

81
# Session persistence (FileAgentSessionStore for local development)
82
session_store = FileAgentSessionStore()
83

84
# To use CosmosDB instead (for horizontal scaling), uncomment below:
85
# from tac_microsoft import CosmosDBAgentSessionStore
86
# session_store = CosmosDBAgentSessionStore(
87
#     endpoint=os.environ["AZURE_COSMOS_ENDPOINT"],
88
#     credential=os.environ["AZURE_COSMOS_KEY"],
89
# )
90

91

92
def on_message(user_message, context, memory_response):
93
    """Customize the user message with context before sending it to the agent."""
94
    prefix = f"[Customer: {context.author_info.address if context.author_info else 'unknown'}]\n"
95
    return prefix + format_memory_context(memory_response, user_message)
96

97

98
def on_error(error, context):
99
    """Return a channel-appropriate error message."""
100
    logger.error("Agent error", extra={"conversation_id": context.conversation_id}, exc_info=error)
101
    if context.channel == "voice":
102
        return "I'm sorry, I'm having trouble right now. Please try again."
103
    return "Sorry, something went wrong. Please try again or call us for help."
104

105

106
async def handle_conversation_ended(context: ConversationSession) -> None:
107
    """Clean up sessions when a conversation closes."""
108
    await session_store.delete(context.conversation_id)
109
    logger.info("Session cleaned up", extra={"conversation_id": context.conversation_id})
110

111

112
tac.on_conversation_ended(handle_conversation_ended)
113

114
connector = AgentFrameworkConnector(
115
    tac=tac,
116
    create_agent=create_agent,
117
    on_message=on_message,
118
    on_error=on_error,
119
    voice_config=VoiceChannelConfig(memory_mode="never"),
120
    sms_config=SMSChannelConfig(memory_mode="always"),
121
    session_store=session_store,
122
)
123

124
server = TACFastAPIServer(
125
    tac=tac,
126
    voice_channel=connector.voice_channel,
127
    messaging_channels=[connector.sms_channel],
128
)
129

130
if __name__ == "__main__":
131
    server.start()

1
AZURE_VOICE_LIVE_ENDPOINT=your-resource.services.ai.azure.com
2
AZURE_VOICE_LIVE_API_KEY=your_voice_live_api_key
3
AZURE_VOICE_LIVE_MODEL=gpt-4o

1
# Fix SSL certificate verification (must be before other imports)
2
import truststore
3
truststore.inject_into_ssl()
4

5
import os
6

7
from dotenv import load_dotenv
8
from tac.tools.base import function_tool
9

10
from tac_microsoft import (
11
    TAC,
12
    TACConfig,
13
    TACFastAPIServer,
14
    VoiceLiveConnector,
15
    VoiceLiveConfig,
16
)
17

18
load_dotenv()
19

20
# Custom tools — use @function_tool to create TACTool instances
21
@function_tool()
22
def look_up_outage(zip_code: str) -> str:
23
    """Check if there is a recent internet outage in a specific zip code.
24

25
    Args:
26
        zip_code: The zip code to check for outages.
27
    """
28
    return f"No reported outages in {zip_code}. Service is operating normally."
29

30

31
tac = TAC(config=TACConfig.from_env())
32

33
config = VoiceLiveConfig(
34
    endpoint=os.environ["AZURE_VOICE_LIVE_ENDPOINT"],
35
    model=os.environ.get("AZURE_VOICE_LIVE_MODEL", "gpt-4o"),
36
    api_key=os.environ.get("AZURE_VOICE_LIVE_API_KEY"),
37
    instructions="You are a helpful customer service agent. Be concise and friendly.",
38
    tools=[look_up_outage],
39
)
40

41
connector = VoiceLiveConnector(
42
    tac=tac,
43
    config=config,
44
)
45

46
server = TACFastAPIServer(
47
    tac=tac,
48
    voice_channel=connector.voice_channel,
49
)
50

51
if __name__ == "__main__":
52
    server.start()