McpToolRegistrationService Classe

Servizio per la gestione di strumenti e server MCP per un agente

Costruttore

McpToolRegistrationService()

Parametri

Nome Descrizione
logger
Valore predefinito: None

Metodi

__init__

Inizializzare il servizio di registrazione dello strumento MCP per OpenAI.
__new__
add_tool_servers_to_agent

Aggiungere nuovi server MCP all'agente creando una nuova istanza di Agent.

Nota: a causa delle limitazioni di OpenAI Agents SDK, i server MCP devono essere impostati durante la creazione dell'agente. Se vengono trovati nuovi server, questo metodo crea una nuova istanza di Agent con tutti i server MCP (esistenti + nuovi) inizializzati correttamente.
cleanup_all_servers

Pulire tutti i server MCP connessi
send_chat_history

Estrarre la cronologia delle chat da una sessione OpenAI e inviarla alla piattaforma MCP.

Questo metodo estrae i messaggi da un oggetto Sessione OpenAI usando get_items() e li invia alla piattaforma MCP per la protezione dalle minacce in tempo reale.
send_chat_history_messages

Inviare messaggi della cronologia delle chat OpenAI alla piattaforma MCP per la protezione dalle minacce.

Questo metodo accetta un elenco di messaggi OpenAI TResponseInputItem, li converte in formato ChatHistoryMessage e li invia alla piattaforma MCP.

Note

Anche se i messaggi sono vuoti o tutti i messaggi vengono filtrati durante la conversione,

la richiesta verrà comunque inviata alla piattaforma MCP. In questo modo si garantisce all'utente

il messaggio di turn_context.activity.text è registrato correttamente per

protezione dalle minacce in tempo reale.

__init__

Inizializzare il servizio di registrazione dello strumento MCP per OpenAI.

__init__(logger: Logger | None = None)

Parametri

Nome Descrizione
logger
Logger | None

Istanza del logger per le operazioni di registrazione.

Valore predefinito: None

__new__ 

__new__(**kwargs)

add_tool_servers_to_agent

Aggiungere nuovi server MCP all'agente creando una nuova istanza di Agent.

Nota: a causa delle limitazioni di OpenAI Agents SDK, i server MCP devono essere impostati durante la creazione dell'agente. Se vengono trovati nuovi server, questo metodo crea una nuova istanza di Agent con tutti i server MCP (esistenti + nuovi) inizializzati correttamente.

async add_tool_servers_to_agent(agent: Agent, auth: Authorization, auth_handler_name: str, context: TurnContext, auth_token: str | None = None) -> Agent

Parametri

Nome Descrizione
agent
Necessario
<xref:agents.agent.Agent>

Agente esistente a cui aggiungere server
auth
Necessario
Authorization

Gestore di autorizzazione per lo scambio di token.
auth_handler_name
Necessario
str

Nome del gestore di autorizzazione.
context
Necessario
TurnContext

Contesto di turno per l'operazione corrente.
auth_token
str | None

Token di autenticazione per accedere ai server MCP.

Valore predefinito: None

Valori restituiti

Tipo Descrizione
<xref:agents.agent.Agent>

Nuova istanza di Agent con tutti i server MCP o l'agente originale se non sono presenti nuovi server

cleanup_all_servers

Pulire tutti i server MCP connessi

async cleanup_all_servers() -> None

Valori restituiti

Tipo Descrizione
None

send_chat_history

Estrarre la cronologia delle chat da una sessione OpenAI e inviarla alla piattaforma MCP.

Questo metodo estrae i messaggi da un oggetto Sessione OpenAI usando get_items() e li invia alla piattaforma MCP per la protezione dalle minacce in tempo reale.

async send_chat_history(turn_context: TurnContext, session: Session, limit: int | None = None, options: ToolOptions | None = None) -> OperationResult

Parametri

Nome Descrizione
turn_context
Necessario
TurnContext

TurnContext da Agents SDK contenente informazioni di conversazione. Deve avere un'attività valida con conversation.id, activity.id e activity.text.
session
Necessario
<xref:agents.memory.session.Session>

Istanza della sessione OpenAI da cui estrarre i messaggi. Deve supportare il metodo get_items() che restituisce un elenco di TResponseInputItem.
limit
int | None

Numero massimo facoltativo di elementi da recuperare dalla sessione. Se Nessuno, recupera tutti gli elementi.

Valore predefinito: None
options
ToolOptions | None

Facoltativo ToolOptions per la personalizzazione. Se non specificato, usa le opzioni predefinite con orchestrator_name="OpenAI".

Valore predefinito: None

Valori restituiti

Tipo Descrizione
OperationResult

OperationResult indica l'esito positivo o negativo. In caso di esito positivo, restituisce OperationResult.success(). In caso di errore, restituisce OperationResult.failed() con i dettagli dell'errore.

Eccezioni

Tipo Descrizione
ValueError

Se turn_context è Nessuno o sessione è Nessuno.

Esempio


>>> from agents import Agent, Runner
>>> from microsoft_agents_a365.tooling.extensions.openai import (
...     McpToolRegistrationService
... )
>>>
>>> service = McpToolRegistrationService()
>>> agent = Agent(name="my-agent", model="gpt-4")
>>>
>>> # In your agent handler:
>>> async with Runner.run(agent, messages) as result:
...     session = result.session
...     op_result = await service.send_chat_history(
...         turn_context, session
...     )
...     if op_result.succeeded:
...         print("Chat history sent successfully")

send_chat_history_messages

Inviare messaggi della cronologia delle chat OpenAI alla piattaforma MCP per la protezione dalle minacce.

Questo metodo accetta un elenco di messaggi OpenAI TResponseInputItem, li converte in formato ChatHistoryMessage e li invia alla piattaforma MCP.

Note

Anche se i messaggi sono vuoti o tutti i messaggi vengono filtrati durante la conversione,

la richiesta verrà comunque inviata alla piattaforma MCP. In questo modo si garantisce all'utente

il messaggio di turn_context.activity.text è registrato correttamente per

protezione dalle minacce in tempo reale.

async send_chat_history_messages(turn_context: TurnContext, messages: List[EasyInputMessageParam | Message | ResponseOutputMessageParam | ResponseFileSearchToolCallParam | ResponseComputerToolCallParam | ComputerCallOutput | ResponseFunctionWebSearchParam | ResponseFunctionToolCallParam | FunctionCallOutput | ToolSearchCall | ResponseToolSearchOutputItemParamParam | ResponseReasoningItemParam | ResponseCompactionItemParamParam | ImageGenerationCall | ResponseCodeInterpreterToolCallParam | LocalShellCall | LocalShellCallOutput | ShellCall | ShellCallOutput | ApplyPatchCall | ApplyPatchCallOutput | McpListTools | McpApprovalRequest | McpApprovalResponse | McpCall | ResponseCustomToolCallOutputParam | ResponseCustomToolCallParam | ItemReference], options: ToolOptions | None = None) -> OperationResult

Parametri

Nome Descrizione
turn_context
Necessario
TurnContext

TurnContext da Agents SDK contenente informazioni di conversazione. Deve avere un'attività valida con conversation.id, activity.id e activity.text.
messages
Necessario
List[<xref:openai.types.responses.easy_input_message_param.EasyInputMessageParam> | <xref:openai.types.responses.response_input_item_param.Message> | <xref:openai.types.responses.response_output_message_param.ResponseOutputMessageParam> | <xref:openai.types.responses.response_file_search_tool_call_param.ResponseFileSearchToolCallParam> | <xref:openai.types.responses.response_computer_tool_call_param.ResponseComputerToolCallParam> | <xref:openai.types.responses.response_input_item_param.ComputerCallOutput> | <xref:openai.types.responses.response_function_web_search_param.ResponseFunctionWebSearchParam> | <xref:openai.types.responses.response_function_tool_call_param.ResponseFunctionToolCallParam> | <xref:openai.types.responses.response_input_item_param.FunctionCallOutput> | <xref:openai.types.responses.response_input_item_param.ToolSearchCall> | <xref:openai.types.responses.response_tool_search_output_item_param_param.ResponseToolSearchOutputItemParamParam> | <xref:openai.types.responses.response_reasoning_item_param.ResponseReasoningItemParam> | <xref:openai.types.responses.response_compaction_item_param_param.ResponseCompactionItemParamParam> | <xref:openai.types.responses.response_input_item_param.ImageGenerationCall> | <xref:openai.types.responses.response_code_interpreter_tool_call_param.ResponseCodeInterpreterToolCallParam> | <xref:openai.types.responses.response_input_item_param.LocalShellCall> | <xref:openai.types.responses.response_input_item_param.LocalShellCallOutput> | <xref:openai.types.responses.response_input_item_param.ShellCall> | <xref:openai.types.responses.response_input_item_param.ShellCallOutput> | <xref:openai.types.responses.response_input_item_param.ApplyPatchCall> | <xref:openai.types.responses.response_input_item_param.ApplyPatchCallOutput> | <xref:openai.types.responses.response_input_item_param.McpListTools> | <xref:openai.types.responses.response_input_item_param.McpApprovalRequest> | <xref:openai.types.responses.response_input_item_param.McpApprovalResponse> | <xref:openai.types.responses.response_input_item_param.McpCall> | <xref:openai.types.responses.response_custom_tool_call_output_param.ResponseCustomToolCallOutputParam> | <xref:openai.types.responses.response_custom_tool_call_param.ResponseCustomToolCallParam> | <xref:openai.types.responses.response_input_item_param.ItemReference>]

Elenco dei messaggi OpenAI TResponseInputItem da inviare. Supporta userMessage, AssistantMessage, SystemMessage e altri tipi di messaggio OpenAI. Può essere vuoto: la richiesta verrà comunque inviata per registrare il messaggio utente da turn_context.activity.text.
options
ToolOptions | None

Facoltativo ToolOptions per la personalizzazione. Se non specificato, usa le opzioni predefinite con orchestrator_name="OpenAI".

Valore predefinito: None

Valori restituiti

Tipo Descrizione
OperationResult

OperationResult indica l'esito positivo o negativo. In caso di esito positivo, restituisce OperationResult.success(). In caso di errore, restituisce OperationResult.failed() con i dettagli dell'errore.

Eccezioni

Tipo Descrizione
ValueError

Se turn_context è Nessuno o i messaggi sono Nessuno.

Esempio


>>> from microsoft_agents_a365.tooling.extensions.openai import (
...     McpToolRegistrationService
... )
>>>
>>> service = McpToolRegistrationService()
>>> messages = [
...     {"role": "user", "content": "Hello"},
...     {"role": "assistant", "content": "Hi there!"},
... ]
>>>
>>> result = await service.send_chat_history_messages(
...     turn_context, messages
... )
>>> if result.succeeded:
...     print("Chat history sent successfully")