01Split AgentApplication routes and rank the catch-all with RouteRank.LASTVerified
AgentApplication in the Microsoft 365 Agents SDK evaluates routes in order and, by default, runs only the first match. Give the catch-all message handler RouteRank.LAST so fixed commands and regex routes are tried first.
Approach and steps
- Register fixed commands with message("/help") and commands with arguments with message(re.compile(...)).
- Pass rank=RouteRank.LAST to the catch-all activity("message") route. Registered first without a rank, it also took /help in our local test.
- String routes were case-sensitive: /HELP fell through to the catch-all. Use a regex with re.IGNORECASE when case should not matter.
- Register a function taking (context, error) with error(). Send users a fixed message and write details to the log.
- Keep route registration in register_routes(). A test can build the same app and feed messages through on_turn().
PythonRuns as a service
"""agent_routes.py: command routes, a ranked catch-all and an error handler for an AgentApplication."""
import re
from microsoft_agents.hosting.core import AgentApplication, RouteRank, TurnContext, TurnState
HELP = "Commands: /help, status <request id> (for example: status R-2026-002)."
STATUS_RE = re.compile(r"^status\s+(R-\d{4}-\d{3})$", re.IGNORECASE)
async def on_help(context: TurnContext, _state: TurnState):
await context.send_activity(HELP)
async def on_status(context: TurnContext, _state: TurnState):
request_id = STATUS_RE.match(context.activity.text.strip()).group(1).upper()
await context.send_activity(f"Looking up {request_id} ...")
async def on_other(context: TurnContext, _state: TurnState):
await context.send_activity("I did not recognise that. Type /help for commands.")
async def on_error(context: TurnContext, error: Exception):
print(f"turn error: {type(error).__name__}") # write details to your log, not to the chat
await context.send_activity("Something went wrong. Please try again later.")
def register_routes(app: AgentApplication) -> AgentApplication:
app.conversation_update("membersAdded")(on_help)
app.message("/help")(on_help)
app.message(STATUS_RE)(on_status)
app.activity("message", rank=RouteRank.LAST)(on_other) # catch-all, evaluated after the routes above
app.error(on_error)
return appOutput（2026-09-12）
'/help' -> ['Commands: /help, status <request id> (for example: status R-2026-002).']
'/HELP' -> ['I did not recognise that. Type /help for commands.']
'status r-2026-002' -> ['Looking up R-2026-002 ...']
'hello' -> ['I did not recognise that. Type /help for commands.']
catch-all registered first with LAST: ['Commands: /help, status <request id> (for example: status R-2026-002).']
catch-all registered first without rank: ['I did not recognise that. Type /help for commands.']Example instruction for Copilot
For an AgentApplication in the Microsoft 365 Agents SDK for Python, write a function that registers three routes: /help, status <request id> (case-insensitive regex) and a catch-all. Register the catch-all with RouteRank.LAST and add an error handler.
Caution
The docs' table lists RouteRank.First, Unspecified and Last, but the Python package 1.5.0 defines FIRST, DEFAULT and LAST. The docs' Python sample uses turn_error, which 1.5.0 does not have; error() works. Names can change between releases, so check them in the version you install. Do not log users' full messages.
Supporting passages from the sources
By default, evaluation stops at the first matching route. Use RouteRank.Last for a catch-all fallback that handles anything not matched by a more specific route.
When an activity arrives, the system evaluates routes in order until it finds a match. By default, only one route runs.
The packages should target Python 3.10 or greater
02Read agent connection settings from environment variables and host them on aiohttpVerified
The Python Agents SDK reads environment variables that start with CONNECTIONS__SERVICE_CONNECTION__SETTINGS__ through load_configuration_from_env() and passes them to MsalConnectionManager. Check the settings before start-up and serve /api/messages behind JWT validation.
Approach and steps
- Name settings CONNECTIONS__<connection name>__SETTINGS__<property>. A connection named SERVICE_CONNECTION is required.
- Choose the authentication flow with AUTHTYPE (Certificate, UserManagedIdentity, SystemManagedIdentity and others). Avoid client secrets.
- Set ANONYMOUS_ALLOWED=True only for local testing. In production, keep JWT validation on.
- Use check_settings() to stop start-up when settings are missing, anonymous access is exposed, or a secret is used.
- Add jwt_authorization_middleware to the aiohttp Application and hand POST /api/messages to start_agent_process().
PythonRuns as a service
"""app.py: build an Agents SDK agent from environment variables and serve it with aiohttp."""
import os
from aiohttp.web import Application, Request, Response, run_app
from microsoft_agents.activity import load_configuration_from_env
from microsoft_agents.authentication.msal import MsalConnectionManager
from microsoft_agents.hosting.aiohttp import CloudAdapter, jwt_authorization_middleware, start_agent_process
from microsoft_agents.hosting.core import AgentApplication, Authorization, MemoryStorage, TurnContext, TurnState
PREFIX = "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__"
def check_settings(env):
"""Return problems that should stop start-up."""
keys = {k[len(PREFIX):] for k in env if k.startswith(PREFIX)}
anonymous = str(env.get(PREFIX + "ANONYMOUS_ALLOWED", "")).lower() == "true"
problems = []
if not keys:
problems.append("no SERVICE_CONNECTION settings")
if anonymous and env.get("AGENT_HOST", "localhost") not in ("localhost", "127.0.0.1"):
problems.append("ANONYMOUS_ALLOWED is for local development only")
if not anonymous and keys and "CLIENTID" not in keys and env.get(PREFIX + "AUTHTYPE") != "SystemManagedIdentity":
problems.append("CLIENTID is missing")
if "CLIENTSECRET" in keys:
problems.append("client secret configured: use Certificate or a managed identity instead")
return problems
def build_agent(env):
config = load_configuration_from_env(env)
storage = MemoryStorage()
connections = MsalConnectionManager(**config)
agent = AgentApplication[TurnState](
storage=storage,
adapter=CloudAdapter(connection_manager=connections),
authorization=Authorization(storage, connections, **config),
**config,
)
@agent.activity("message")
async def on_message(context: TurnContext, _state: TurnState):
await context.send_activity(f"you said: {context.activity.text}")
return agent, connections
def build_web_app(agent, connections):
async def entry_point(req: Request) -> Response:
return await start_agent_process(req, agent, agent.adapter)
web = Application(middlewares=[jwt_authorization_middleware])
web.router.add_post("/api/messages", entry_point)
web["agent_configuration"] = connections.get_default_connection_configuration()
web["agent_app"] = agent
web["adapter"] = agent.adapter
return web
if __name__ == "__main__":
issues = check_settings(os.environ)
if issues:
raise SystemExit("stopped: " + "; ".join(issues))
agent, connections = build_agent(os.environ)
run_app(build_web_app(agent, connections), host=os.environ.get("AGENT_HOST", "localhost"),
port=int(os.environ.get("PORT", 3978)))Output（2026-09-12）
empty -> ['no SERVICE_CONNECTION settings']
local anonymous -> ok
anonymous on a public host -> ['ANONYMOUS_ALLOWED is for local development only']
certificate -> ok
system managed identity -> ok
secret -> ['client secret configured: use Certificate or a managed identity instead']
['POST /api/messages']
no settings -> No service connection configuration provided.Example instruction for Copilot
Write an app.py that runs a Microsoft 365 Agents SDK (Python) agent on aiohttp. Read settings with load_configuration_from_env(os.environ) and build MsalConnectionManager and CloudAdapter. Add a function that checks, before start-up, that SERVICE_CONNECTION exists, that anonymous access is only used locally, and that no secret is configured.
Caution
With no connection settings at all, MsalConnectionManager raised ValueError. The quickstart's bare CloudAdapter() also raised ValueError in 1.5.0; it needed a connection manager. ANONYMOUS_ALLOWED is for local development only; never use it where the agent is reachable from outside. Keep .env files out of source control.
Supporting passages from the sources
The agent obtains MSAL configuration at runtime from environment variables using the helper function load_configuration_from_env().
Connection settings use the format CONNECTIONS__<CONNECTION_NAME>__SETTINGS__<PROPERTY>.
The connection manager requires at minimum a connection named SERVICE_CONNECTION.
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__ANONYMOUS_ALLOWED=True
For production environments, ensure JWT validation is enabled through the CloudAdapter configuration.
03Build a lookup API for an agent as an HTTP-triggered Azure FunctionVerified
With the Python v2 programming model, write an HTTP function that takes a request ID and returns its ledger status as JSON. Set the auth level to FUNCTION so callers must send an access key in the x-functions-key header.
Approach and steps
- Create func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION) and register the function with @app.route(route=..., methods=["GET"]).
- Read input from req.params. If it is missing, return status 400 with the reason as JSON.
- Return unknown IDs as JSON with found: false too, so the agent can phrase its reply.
- Fix the returned fields to English keys and reuse the same names in the OpenAPI file for the REST API tool.
- Read the data location from app settings, for example os.getenv("LEDGER_PATH"), instead of hard-coding it.
- An HTTP function must respond within 230 seconds. For long work, return an acknowledgement and finish later.
PythonRuns as a service
"""function_app.py: HTTP-triggered lookup that an agent or a Copilot Studio REST API tool calls."""
import csv
import json
import logging
import os
import azure.functions as func
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
COLUMNS = {"文書名": "title", "種別": "doc_type", "状態": "state", "指摘件数": "findings", "完了日": "completed"}
def load_ledger(path):
with open(path, encoding="utf-8-sig", newline="") as f:
return {row["依頼ID"]: row for row in csv.DictReader(f)}
def json_response(body, status_code=200):
return func.HttpResponse(json.dumps(body, ensure_ascii=False), status_code=status_code,
mimetype="application/json")
@app.route(route="request_status", methods=["GET"])
def request_status(req: func.HttpRequest) -> func.HttpResponse:
request_id = (req.params.get("id") or "").strip().upper()
if not request_id:
return json_response({"error": "query parameter 'id' is required"}, 400)
row = load_ledger(os.getenv("LEDGER_PATH", "ledger.csv")).get(request_id)
if row is None:
return json_response({"id": request_id, "found": False})
logging.info("request_status looked up %s", request_id)
result = {"id": request_id, "found": True}
result.update({en: row[ja] for ja, en in COLUMNS.items()})
result["findings"] = int(result["findings"] or 0)
return json_response(result)Output（2026-09-12）
200 {"id": "R-2026-002", "found": true, "title": "文書02（架空）", "doc_type": "プレスリリース", "state": "受付", "findings": 9, "completed": ""}
200 {"id": "R-2099-001", "found": false}
400 {"error": "query parameter 'id' is required"}Example instruction for Copilot
Using the Azure Functions Python v2 model, write a GET /api/request_status?id=... function that returns the matching ledger row from a CSV file as JSON with English keys. Use auth level FUNCTION, return 400 when id is missing and found: false when nothing matches.
Caution
An HTTP-triggered function has 230 seconds to respond, whatever the function timeout is. With the FUNCTION auth level, anyone who knows the key can call it, so keep keys in Key Vault or app settings, never in code or the OpenAPI file. Core Tools does not enforce authorization locally, so test keys in Azure. Return only the fields the task needs.
Supporting passages from the sources
You can include the access key in the URL by using the ?code= query string or in the request header (x-functions-key).
Regardless of the function app timeout setting, 230 seconds is the maximum amount of time that an HTTP triggered function can take to respond to a request.
Determines what keys, if any, need to be present on the request in order to invoke the function.
By default, authorization isn't enforced locally for HTTP endpoints.
04Unit-test an HTTP function through build().get_user_function()Verified
In the Python v2 model, call build().get_user_function() on the decorated name to get the plain function and pass it a func.HttpRequest. No Core Tools or network is needed. From azure-functions 1.21.0 you can also call the function directly.
Approach and steps
- Build a test request with func.HttpRequest(method=..., url=..., body=..., params=...).
- Pass it to the function returned by name.build().get_user_function() and check the status code and body.
- The docs' example imports my_function but calls main.build(). Use the name of your decorated function instead.
- Test broken JSON and missing fields as well as valid input.
- With azure-functions 1.21.0 or later, calling count_findings(req) directly gave the same result in our local venv.
- Run the tests with unittest or pytest before every deployment.
PythonRuns locally
"""Unit-test a Python v2 HTTP function without Core Tools (function and tests in one file for brevity)."""
import json
import sys
import unittest
import azure.functions as func
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
@app.route(route="count_findings", methods=["POST"])
def count_findings(req: func.HttpRequest) -> func.HttpResponse:
try:
rows = req.get_json().get("rows", [])
except ValueError:
return func.HttpResponse("JSON body is required", status_code=400)
total = sum(int(r.get("findings", 0)) for r in rows)
return func.HttpResponse(json.dumps({"rows": len(rows), "findings": total}), mimetype="application/json")
def make_request(body: bytes) -> func.HttpRequest:
return func.HttpRequest(method="POST", url="/api/count_findings", body=body)
class CountFindingsTest(unittest.TestCase):
def call(self, body):
return count_findings.build().get_user_function()(make_request(body))
def test_sum(self):
resp = self.call(json.dumps({"rows": [{"findings": 2}, {"findings": "3"}]}).encode())
self.assertEqual(resp.status_code, 200)
self.assertEqual(json.loads(resp.get_body()), {"rows": 2, "findings": 5})
def test_missing_rows(self):
self.assertEqual(json.loads(self.call(b"{}").get_body()), {"rows": 0, "findings": 0})
def test_bad_json(self):
self.assertEqual(self.call(b"not json").status_code, 400)
def test_direct_call(self):
resp = count_findings(make_request(json.dumps({"rows": [{"findings": 4}]}).encode()))
self.assertEqual(json.loads(resp.get_body())["findings"], 4)
if __name__ == "__main__":
suite = unittest.defaultTestLoader.loadTestsFromTestCase(CountFindingsTest)
result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
print("tests passed" if result.wasSuccessful() else "tests failed")
sys.exit(0 if result.wasSuccessful() else 1)Output（2026-09-12）
test_bad_json (__main__.CountFindingsTest.test_bad_json) ... ok
test_direct_call (__main__.CountFindingsTest.test_direct_call) ... ok
test_missing_rows (__main__.CountFindingsTest.test_missing_rows) ... ok
test_sum (__main__.CountFindingsTest.test_sum) ... ok
----------------------------------------------------------------------
Ran 4 tests in 0.001s
OK
tests passedExample instruction for Copilot
Write unittest tests for this Azure Functions Python v2 HTTP function. Build requests with func.HttpRequest and call name.build().get_user_function(). Cover valid input, broken JSON and missing fields.
Caution
These tests cover only the logic inside the function. Auth level, routes and host.json settings are not exercised, so also check them with func start. The docs' unit-test example uses a variable name that does not match its import and will not run as written.
Supporting passages from the sources
By using azure-functions >= 1.21.0, you can also call functions directly by using the Python interpreter without running Core Tools.
func_call = main.build().get_user_function()
For most bindings, you can create a mock input object by creating an instance of an appropriate class from the azure.functions package.
05Generate an OpenAPI v2 file for a Copilot Studio REST API toolVerified
Copilot Studio REST API tools are created from an OpenAPI v2 JSON file. Keep the function's inputs, outputs and descriptions in a Python dict, check that descriptions are filled in, write out_openapi.json and upload it when adding the tool.
Approach and steps
- In the agent's [Tools], select [Add a tool] > [New tool] > [REST API] and upload the generated JSON.
- Set swagger to "2.0". A v3 file is translated to v2 automatically.
- Write specific descriptions for the API and each operation, saying when to use it. The agent selects tools from these descriptions.
- Describe every parameter and response field too. Blank descriptions block the wizard.
- For authentication choose [API key], with x-functions-key as the Parameter name and Header as the Parameter location.
- Leave out operations users should not run (updates, deletes), or deselect them when choosing tools.
PythonRuns locally
"""Write an OpenAPI 2.0 (swagger) file for the request_status function, for a Copilot Studio REST API tool."""
import json
import os
HOST = os.environ.get("FUNCTION_HOST", "localhost:7071") # set to your function app's host name
SPEC = {
"swagger": "2.0",
"info": {
"title": "Request status",
"version": "1.0.0",
"description": "Looks up a document review request by its ID (for example R-2026-002). Use it when a user "
"asks about the state, number of findings or completion date of a request.",
},
"host": HOST,
"basePath": "/api",
"schemes": ["http" if HOST.startswith("localhost") else "https"],
"produces": ["application/json"],
"securityDefinitions": {"functionKey": {"type": "apiKey", "name": "x-functions-key", "in": "header"}},
"security": [{"functionKey": []}],
"paths": {
"/request_status": {
"get": {
"operationId": "GetRequestStatus",
"summary": "Get request status",
"description": "Returns title, document type, state, number of findings and completion date "
"for one request ID. Returns found=false when the ID is not in the ledger.",
"parameters": [{"name": "id", "in": "query", "required": True, "type": "string",
"description": "Request ID such as R-2026-002"}],
"responses": {
"200": {"description": "Lookup result", "schema": {"type": "object", "properties": {
"id": {"type": "string", "description": "Request ID"},
"found": {"type": "boolean", "description": "False when the ID is unknown"},
"title": {"type": "string", "description": "Document title"},
"doc_type": {"type": "string", "description": "Document type"},
"state": {"type": "string", "description": "Current review state"},
"findings": {"type": "integer", "description": "Number of findings"},
"completed": {"type": "string", "description": "Completion date, empty if open"}}}},
"400": {"description": "The id parameter is missing"},
},
}
}
},
}
def check(spec):
problems = []
if spec.get("swagger") != "2.0":
problems.append("REST API tools need an OpenAPI v2 (swagger 2.0) file")
if len(spec["info"].get("description", "")) < 40:
problems.append("API description is too short for tool selection")
for path, ops in spec["paths"].items():
for method, op in ops.items():
where = f"{method.upper()} {path}"
if len(op.get("description", "")) < 40:
problems.append(f"{where}: operation description is too short")
for p in op.get("parameters", []):
if not p.get("description"):
problems.append(f"{where}: parameter {p['name']} has no description")
for code, resp in op.get("responses", {}).items():
for name, prop in resp.get("schema", {}).get("properties", {}).items():
if not prop.get("description"):
problems.append(f"{where}: response {code} field {name} has no description")
return problems
if __name__ == "__main__":
issues = check(SPEC)
with open("out_openapi.json", "w", encoding="utf-8") as f:
json.dump(SPEC, f, ensure_ascii=False, indent=2)
print("wrote out_openapi.json; operations:", sum(len(v) for v in SPEC["paths"].values()))
print("issues:", issues or "none")Output（2026-09-12）
wrote out_openapi.json; operations: 1
issues: noneExample instruction for Copilot
Write a Python script that turns the Azure Functions endpoint GET /api/request_status?id=... into an OpenAPI 2.0 (swagger) JSON file that a Copilot Studio REST API tool can load. Add a function that lists operations or parameters with missing descriptions.
Caution
REST API tools are in preview, and the docs say preview features are not meant for production use. Do not put key values in the file; enter them when you create the connection. Data policies in your environment may block HTTP features. Any operation in the file can be called by the agent on its own judgement.
Supporting passages from the sources
You must create REST API tools from an OpenAPI v2 specification. This requirement is due to the behavior of Power Platform in processing API specifications.
Provide a detailed description, because your agent orchestration uses the description to determine when to use the particular tool.
Parameter location: How you send the key for the API. Select either Header or Query.
If any of the descriptions are blank, you must complete them before you can move forward.
Preview features aren't meant for production use and may have restricted functionality.
You can include the access key in the URL by using the ?code= query string or in the request header (x-functions-key).
06Call an Azure Functions API from an agent handler and handle timeoutsVerified
Inside an Agents SDK route handler, call the function's API with GET through aiohttp's ClientSession. Read the URL and key from environment variables, and send users a fixed message on timeouts or connection failures.
Approach and steps
- Read the API URL and key from app settings, for example STATUS_API_URL and STATUS_API_KEY. Locally, default to Core Tools at http://localhost:7071.
- Send the key in the x-functions-key header.
- Set an upper bound with ClientTimeout(total=...), shorter than the function-side limit of 230 seconds, so users are not left waiting.
- Catch ClientError and TimeoutError, tell users to try again later, and log the cause.
- Keep the reply formatting in format_reply() so it can be unit-tested on its own.
- Register the handler on a route, for example AGENT_APP.message(STATUS_RE)(on_status).
PythonRuns as a service
"""Agent handler that calls the request_status function and replies with the result."""
import os
import re
from aiohttp import ClientError, ClientSession, ClientTimeout
from microsoft_agents.hosting.core import TurnContext, TurnState
STATUS_RE = re.compile(r"^status\s+(R-\d{4}-\d{3})$", re.IGNORECASE)
TIMEOUT = ClientTimeout(total=float(os.environ.get("STATUS_API_TIMEOUT", "20")))
async def fetch_status(request_id):
url = os.environ.get("STATUS_API_URL", "http://localhost:7071/api/request_status")
headers = {}
function_key = os.environ.get("STATUS_API_KEY") # from app settings or Key Vault, never from code
if function_key:
headers["x-functions-key"] = function_key
async with ClientSession(timeout=TIMEOUT) as session:
async with session.get(url, params={"id": request_id}, headers=headers) as resp:
resp.raise_for_status()
return await resp.json()
def format_reply(data):
if not data.get("found"):
return f"{data.get('id')}: not found in the ledger."
done = data.get("completed") or "-"
return (f"{data['id']} {data['title']} ({data['doc_type']}): {data['state']}, "
f"findings {data['findings']}, completed {done}")
async def on_status(context: TurnContext, _state: TurnState):
request_id = STATUS_RE.match(context.activity.text.strip()).group(1).upper()
try:
data = await fetch_status(request_id)
except (ClientError, TimeoutError) as error:
print(f"status lookup failed: {type(error).__name__}")
await context.send_activity("The status service did not respond. Please try again later.")
return
await context.send_activity(format_reply(data))
# Register it on your AgentApplication, for example: AGENT_APP.message(STATUS_RE)(on_status)Output（2026-09-12）
R-2026-002 文書02（架空） (プレスリリース): 受付, findings 9, completed -
R-2026-999: not found in the ledger.
header sent: True
status lookup failed: ClientConnectorError
service down -> The status service did not respond. Please try again later.Example instruction for Copilot
Write a Microsoft 365 Agents SDK (Python) route handler that calls the Azure Functions endpoint GET /api/request_status with aiohttp. Read the URL and key from environment variables, send the key in the x-functions-key header, set a timeout, and reply with a fixed message on failure.
Caution
A function key is shared by every caller, so access is not limited per user. Limit what the function returns to what the task needs. Keep the key in Key Vault or app settings and never log it. Validate the response shape before showing it to users.
Supporting passages from the sources
You can include the access key in the URL by using the ?code= query string or in the request header (x-functions-key).
Regardless of the function app timeout setting, 230 seconds is the maximum amount of time that an HTTP triggered function can take to respond to a request.
http_trigger: http://localhost:7071/api/http_trigger
07Ask a Copilot Studio agent from Python with the Copilot Studio client libraryVerified
microsoft-agents-copilotstudio-client connects Python code directly to agents built in Copilot Studio. Create a CopilotClient with a token from a signed-in user, start a conversation and send questions.
Approach and steps
- Publish the agent in Copilot Studio and note Schema name and Environment Id under [Settings] > [Advanced] > [Metadata].
- Register a public client (native) app in Entra ID and add the delegated Power Platform API permission CopilotStudio.Copilots.Invoke.
- Put the environment ID, schema name, app ID and tenant ID in environment variables starting with COPILOTSTUDIOAGENT__.
- Get a user token for https://api.powerplatform.com/.default with MSAL's PublicClientApplication.
- Create ConnectionSettings and CopilotClient, get a conversation ID from start_conversation(), then call ask_question().
PythonSign-in required (Microsoft Entra)
"""Ask a published Copilot Studio agent one question from Python (signed-in user)."""
import asyncio
import os
from msal import PublicClientApplication
from microsoft_agents.activity import ActivityTypes
from microsoft_agents.copilotstudio.client import ConnectionSettings, CopilotClient
SCOPES = ["https://api.powerplatform.com/.default"]
NAMES = ["ENVIRONMENTID", "SCHEMANAME", "AGENTAPPID", "TENANTID"]
def settings_from_env():
values = {n: os.environ.get("COPILOTSTUDIOAGENT__" + n, "") for n in NAMES}
missing = [n for n, v in values.items() if not v]
if missing:
raise SystemExit("set COPILOTSTUDIOAGENT__" + ", COPILOTSTUDIOAGENT__".join(missing))
return values
def acquire_user_token(app_id, tenant_id):
app = PublicClientApplication(client_id=app_id, authority=f"https://login.microsoftonline.com/{tenant_id}")
accounts = app.get_accounts()
result = app.acquire_token_silent(SCOPES, account=accounts[0]) if accounts else None
if not result:
result = app.acquire_token_interactive(scopes=SCOPES)
return result["access_token"]
async def ask_once(question):
v = settings_from_env()
settings = ConnectionSettings(
environment_id=v["ENVIRONMENTID"],
agent_identifier=v["SCHEMANAME"],
cloud=None,
copilot_agent_type=None,
custom_power_platform_cloud=None,
)
client = CopilotClient(settings, acquire_user_token(v["AGENTAPPID"], v["TENANTID"]))
conversation_id = None
async for activity in client.start_conversation(True):
conversation_id = activity.conversation.id
replies = []
async for reply in client.ask_question(question, conversation_id):
if reply.type == ActivityTypes.message and reply.text:
replies.append(reply.text)
return replies
if __name__ == "__main__":
for text in asyncio.run(ask_once(os.environ.get("QUESTION", "What can you help me with?"))):
print(text)Example instruction for Copilot
Using microsoft-agents-copilotstudio-client, write Python code that asks a Copilot Studio agent one question and prints only the reply text. Read IDs from COPILOTSTUDIOAGENT__ environment variables and get the token with MSAL interactive sign-in.
Caution
The client needs a user token; the official sample says server-to-server (S2S) is not supported yet. Agents built on the GitHub Copilot harness are not officially supported. The package is not installed in our venv, so the code was only syntax-checked, not run.
Supporting passages from the sources
The Copilot Studio Client is for connecting to and interacting with agents created in Microsoft Copilot Studio.
Currently, you can only use the Copilot Studio client library with Copilot Studio agents created by using the standard harness.
The CopilotStudio Client requires a User Token to operate.
S2S is not currently supported for Copilot Studio.
"scopes": ["https://api.powerplatform.com/.default"],